fix(io): add websocket scheme props (#3874)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

docs/en_US/guide/sinks/builtin/websocket.md

@@ -7,6 +7,7 @@ The action is used for publishing output message into websocket channel.
 | Property name  | Optional | Description                                           |
 | addr           | false    | The address of the websocket sink server, like: 127.0.0.1:8080 |
 | path           | true    | The url path of the websocket sink server, like: /api/data     |
+| scheme           | true    | The url scheme of the websocket sink server, like: ws or wss    |
 | insecureSkipVerify | false | whether to ignore SSL verification |
 | certificationPath  | true | websocket client ssl verification crt file path |
 | privateKeyPath     | true | Key file path for websocket client SSL verification |

docs/en_US/guide/sources/builtin/websocket.md

@@ -15,6 +15,7 @@ When you need eKuiper as a websocket client, you need to specify the server addr
 ```yaml
 default:
   addr: 127.0.0.1:8080
+  scheme: ws
 ```
 
 ```sql

docs/zh_CN/guide/sinks/builtin/websocket.md

@@ -2,18 +2,19 @@
 
 ## 属性
 
-| 属性名称         | 是否必填 | 说明                                       |
-|--------------|------|------------------------------------------|
-| addr         | 否    | websocket server 的地址，如: 127.0.0.1:8080   |
-| path     | 是    | websocket server 的 url path，如: /api/data |
-| insecureSkipVerify | 否   | 是否忽略 SSL 验证                              |
-| certificationPath  | 是   | websocket 客户端 ssl 验证的 crt 文件路径           |
-| privateKeyPath     | 是   | websocket 客户端 ssl 验证的 key 文件路径               |
-| rootCaPath         | 是   | websocket 客户端 ssl 验证的 ca 证书文件路径              |
-| certficationRaw    | 是   | websocket 客户端 ssl 验证，经过 base64 编码过的 crt 原文,  如果同时定义了 `certificationPath` 将会先用该参数。       |
-| privateKeyRaw      | 是   | websocket 客户端 ssl 验证，经过 base64 编码过的的 key 原文,  如果同时定义了 `privateKeyPath` 将会先用该参数。       |
+| 属性名称               | 是否必填 | 说明                                                                                |
+|--------------------|------|-----------------------------------------------------------------------------------|
+| addr               | 否    | websocket server 的地址，如: 127.0.0.1:8080                                            |
+| path               | 是    | websocket server 的 url path，如: /api/data                                          |
+| scheme             | 是    | websocket server 的 url scheme，如: ws 或者 wss                                        |
+| insecureSkipVerify | 否   | 是否忽略 SSL 验证                                                                       |
+| certificationPath  | 是   | websocket 客户端 ssl 验证的 crt 文件路径                                                    |
+| privateKeyPath     | 是   | websocket 客户端 ssl 验证的 key 文件路径                                                    |
+| rootCaPath         | 是   | websocket 客户端 ssl 验证的 ca 证书文件路径                                                   |
+| certficationRaw    | 是   | websocket 客户端 ssl 验证，经过 base64 编码过的 crt 原文,  如果同时定义了 `certificationPath` 将会先用该参数。 |
+| privateKeyRaw      | 是   | websocket 客户端 ssl 验证，经过 base64 编码过的的 key 原文,  如果同时定义了 `privateKeyPath` 将会先用该参数。   |
 | rootCARaw          | 是   | websocket 客户端 ssl 验证，经过 base64 编码过的的 ca 原文,  如果同时定义了 `rootCAPath` 将会先用该参数。        |
-| checkConnection    | 否 | 是否检查 websocket endpoint 已经存在连接   |
+| checkConnection    | 否 | 是否检查 websocket endpoint 已经存在连接                                                    |
 
 其他通用的 sink 属性也支持，请参阅[公共属性](../overview.md#公共属性)。
 

docs/zh_CN/guide/sources/builtin/websocket.md

@@ -15,6 +15,7 @@ eKuiper 可以作为 websocket 客户端，向远端的 websocket 服务器发
 ```yaml
 default:
   addr: 127.0.0.1:8080
+  scheme: ws
 ```
 
 ```sql

internal/io/http/httpserver/websocketConn.go

@@ -37,7 +37,9 @@ func (w *WebsocketConnection) GetId(ctx api.StreamContext) string {
 }
 
 func (w *WebsocketConnection) Provision(ctx api.StreamContext, conId string, props map[string]any) error {
-	cfg := &wscConfig{}
+	cfg := &wscConfig{
+		Scheme: "ws",
+	}
 	if err := cast.MapToStruct(props, cfg); err != nil {
 		return err
 	}
@@ -64,7 +66,7 @@ func (w *WebsocketConnection) Dial(ctx api.StreamContext) error {
 		if err != nil {
 			return err
 		}
-		c := NewWebsocketClient(w.cfg.Addr, w.cfg.Path, tlsConfig)
+		c := NewWebsocketClient(w.cfg.Scheme, w.cfg.Addr, w.cfg.Path, tlsConfig)
 		if err := c.Connect(); err != nil {
 			return err
 		}
@@ -78,6 +80,7 @@ type wscConfig struct {
 	Path       string `json:"path"`
 	Datasource string `json:"datasource"`
 	Addr       string `json:"addr"`
+	Scheme     string `json:"scheme"`
 }
 
 func (w *WebsocketConnection) Ping(ctx api.StreamContext) error {

internal/io/http/httpserver/websocket_client.go

@@ -32,6 +32,7 @@ type WebsocketClient struct {
 	RecvTopic string
 	SendTopic string
 
+	scheme    string
 	addr      string
 	path      string
 	tlsConfig *tls.Config
@@ -40,8 +41,12 @@ type WebsocketClient struct {
 	cancel    context.CancelFunc
 }
 
-func NewWebsocketClient(addr, path string, tlsConfig *tls.Config) *WebsocketClient {
+func NewWebsocketClient(scheme, addr, path string, tlsConfig *tls.Config) *WebsocketClient {
+	if scheme == "" {
+		scheme = "ws"
+	}
 	return &WebsocketClient{
+		scheme:    scheme,
 		addr:      addr,
 		path:      path,
 		tlsConfig: tlsConfig,
@@ -57,7 +62,7 @@ func (c *WebsocketClient) Connect() error {
 	if len(c.addr) < 1 {
 		return fmt.Errorf("addr should be defined")
 	}
-	u := url.URL{Scheme: "ws", Host: c.addr, Path: c.path}
+	u := url.URL{Scheme: c.scheme, Host: c.addr, Path: c.path}
 	conn, _, err := d.Dial(u.String(), nil)
 	if err != nil {
 		return err

internal/io/http/httpserver/websocket_client_test.go

@@ -31,7 +31,7 @@ func TestWebsocketClient(t *testing.T) {
 		s.Close()
 	}()
 	ctx := mockContext.NewMockContext("1", "2")
-	wc := NewWebsocketClient(s.URL[len("http://"):], "/ws", nil)
+	wc := NewWebsocketClient("ws", s.URL[len("http://"):], "/ws", nil)
 	require.NoError(t, wc.Connect())
 	rt, st := wc.Run(ctx)
 	pubsub.CreatePub(st)