.drone.yml

@@ -9,4 +9,4 @@ steps:
     GO111MODULE: on
     GOPROXY: https://goproxy.cn,direct
   commands:
-  - go build -tags "custom,inputs.cl_104,outputs.influxdb_v2,parsers.cl_104" ./cmd/telegraf
+  - go build -tags "custom,outputs.influxdb_v2" ./cmd/telegraf

etc/telegraf/telegraf.conf

@@ -40,9 +40,9 @@
 
 [[inputs.cl_104]]
   service_address = "tcp://:8899"
-  cl_url="ws://127.0.0.1:8899/api/104"
+  path_cl = "/api/104"
-  api_path="/api/104"
+  path_up = "/api/104up"
   pong_wait = "60s"
   ping_period = "54s"
-  write_wait = "10s"
+  write_waite = "10s"
   data_format = "cl_104"

plugins/inputs/all/cl_104.go

@@ -1,4 +1,4 @@
-//go:build !custom || inputs || inputs.cl_104
+//go:build custom || inputs || inputs.cl_104
 
 package all
 

plugins/inputs/all/cl_kafka_consumer.go

@@ -1,4 +1,4 @@
-//go:build !custom || inputs || inputs.cl_kafka_consumer
+//go:build custom || inputs || inputs.cl_kafka_consumer
 
 package all
 

plugins/inputs/all/cl_kafka_subscriber.go

@@ -1,4 +1,4 @@
-//go:build !custom || inputs || inputs.cl_kafka_subscriber
+//go:build custom || inputs || inputs.cl_kafka_subscriber
 
 package all
 

plugins/inputs/cl_104/README.md

@@ -40,10 +40,9 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   service_address = "tcp://:8080"
   ## service_address = "tcp://:8443"
 
-  ## URL to connect to server
+  ## Paths to listen to.
-  cl_url="ws://127.0.0.1:8899/api/104"
+  # path_cl="/api/104"
-  ## path for access
+  # path_up="/api/104up"
-  api_path="/api/104"
 
   ## maximum duration before timing out read of the request
   # read_timeout = "10s"

plugins/inputs/cl_104/cl_104.go

@@ -32,8 +32,8 @@ type CL104 struct {
 	ServiceAddress string          `toml:"service_address"`
 	ReadTimeout    config.Duration `toml:"read_timeout"`
 	WriteTimeout   config.Duration `toml:"write_timeout"`
-	CLURL          string          `toml:"cl_url"`
+	PathCl         string          `toml:"path_cl"`
-	ApiPath        string          `toml:"api_path"`
+	PathUp         string          `toml:"path_up"`
 	PongWait       config.Duration `toml:"pong_wait"`
 	PingPeriod     config.Duration `toml:"ping_period"`
 	WriteWait      config.Duration `toml:"write_wait"`
@@ -52,6 +52,7 @@ type CL104 struct {
 	listener net.Listener
 	url      *url.URL
 
+	route  map[string]func(res http.ResponseWriter, req *http.Request)
 	upChan chan []byte // confirm
 	clChan chan []byte // command
 
@@ -115,6 +116,10 @@ func (h *CL104) Init() error {
 	h.url = u
 	h.tlsConf = tlsConf
 
+	h.route = map[string]func(res http.ResponseWriter, req *http.Request){
+		h.PathCl: h.serveClstream,
+		h.PathUp: h.serveUpstream,
+	}
 	h.upChan = make(chan []byte, 16)
 	h.clChan = make(chan []byte, 16)
 
@@ -159,9 +164,6 @@ func (h *CL104) Start(acc telegraf.Accumulator) error {
 
 	server := h.createHTTPServer()
 
-	ctx, cancel := context.WithCancel(context.Background())
-	go h.connectingCL(ctx, cancel)
-
 	h.wg.Add(1)
 	go func() {
 		defer h.wg.Done()
@@ -170,7 +172,6 @@ func (h *CL104) Start(acc telegraf.Accumulator) error {
 				h.Log.Errorf("Serve failed: %v", err)
 			}
 			close(h.close)
-			cancel()
 		}
 	}()
 
@@ -192,9 +193,9 @@ func (h *CL104) Stop() {
 
 // ServeHTTP implements [http.Handler]
 func (h *CL104) ServeHTTP(res http.ResponseWriter, req *http.Request) {
-	handler := h.serveUpstream
 
-	if req.URL.Path != h.ApiPath {
+	handler, ok := h.route[req.URL.Path]
+	if !ok {
 		handler = http.NotFound
 	}
 

plugins/inputs/cl_104/clstream.go

@@ -4,39 +4,43 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/influxdata/telegraf/internal"
 )
 
-func (h *CL104) connectingCL(ctx context.Context, cancel context.CancelFunc) {
+var clConnNum int64
-	for {
+
+func (h *CL104) serveClstream(res http.ResponseWriter, req *http.Request) {
 	select {
-		case <-ctx.Done():
-			return
 	case <-h.close:
+		res.WriteHeader(http.StatusGone)
 		return
 	default:
-			h.newConnectCL(ctx, cancel)
+		if atomic.SwapInt64(&clConnNum, 1) > 0 {
-			time.Sleep(time.Second * 5)
+			res.WriteHeader(http.StatusConflict)
-		}
-	}
-}
-
-func (h *CL104) newConnectCL(ctx context.Context, cancel context.CancelFunc) {
-	c, _, err := websocket.DefaultDialer.DialContext(ctx, h.CLURL, nil)
-	if err != nil {
-		h.Log.Error("client dial:", err)
 			return
 		}
-	defer c.Close()
+	}
+
+	conn, err := upgrader.Upgrade(res, req, nil)
+	if err != nil {
+		h.Log.Error(err)
+		return
+	}
+	defer conn.Close()
+
+	stopCtx, stopCancel := context.WithCancel(context.Background())
+	defer stopCancel()
 
 	session := &wsSession{
-		conn:   c,
+		conn:   conn,
-		ctx:    ctx,
+		ctx:    stopCtx,
-		cancel: cancel,
+		cancel: stopCancel,
 		ctrlCh: make(chan wsMsg, 2),
 	}
 
@@ -70,6 +74,7 @@ func (h *CL104) startClWorkers(session *wsSession) {
 	}()
 
 	wg.Wait()
+	atomic.SwapInt64(&clConnNum, 0)
 }
 
 func (h *CL104) monitorClWrite(session *wsSession) {
@@ -133,7 +138,7 @@ func (h *CL104) monitorClRead(session *wsSession) {
 					h.Log.Error(err)
 				}
 			default:
-				h.Log.Info("rm not text")
+				h.Log.Info("not text:", string(rm))
 			}
 		}
 	}

plugins/inputs/cl_104/sample.conf

@@ -6,10 +6,9 @@
   service_address = "tcp://:8080"
   ## service_address = "tcp://:8443"
 
-  ## URL to connect to server
+  ## Paths to listen to.
-  cl_url="ws://127.0.0.1:8899/api/104"
+  # path_cl="/api/104"
-  ## path for access
+  # path_up="/api/104up"
-  api_path="/api/104"
 
   ## maximum duration before timing out read of the request
   # read_timeout = "10s"

plugins/parsers/all/cl_104.go

@@ -1,4 +1,4 @@
-//go:build !custom || parsers || parsers.cl_104
+//go:build custom || parsers || parsers.cl_104
 
 package all
 

plugins/parsers/all/phasor_binary.go

@@ -1,4 +1,4 @@
-//go:build !custom || parsers || parsers.phasor_binary
+//go:build custom || parsers || parsers.phasor_binary
 
 package all