feat(ext): restore zmq source/sink

Signed-off-by: Jiyong Huang <[email protected]>

extensions/impl/zmq/conf.go

@@ -0,0 +1,39 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package zmq
+
+import (
+	"errors"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/cast"
+)
+
+type c struct {
+	Topic  string `json:"datasource"`
+	Server string `json:"server"`
+}
+
+func validate(_ api.StreamContext, props map[string]any) (*c, error) {
+	sc := &c{}
+	err := cast.MapToStruct(props, sc)
+	if err != nil {
+		return nil, err
+	}
+	if sc.Server == "" {
+		return nil, errors.New("missing server address")
+	}
+	return sc, nil
+}

extensions/impl/zmq/conf_test.go

@@ -0,0 +1,66 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package zmq
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		n string
+		p map[string]any
+		c *c
+		e string
+	}{
+		{
+			n: "normal",
+			p: map[string]any{
+				"server":     "tcp://127.0.0.1:5563",
+				"datasource": "t1",
+			},
+			c: &c{
+				Server: "tcp://127.0.0.1:5563",
+				Topic:  "t1",
+			},
+		},
+		{
+			n: "wrong type",
+			p: map[string]any{
+				"server":     "tcp://127.0.0.1:5563",
+				"datasource": 1,
+			},
+			e: "1 error(s) decoding:\n\n* 'datasource' expected type 'string', got unconvertible type 'int', value: '1'",
+		},
+		{
+			n: "missing server",
+			p: map[string]any{},
+			e: "missing server address",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.n, func(t *testing.T) {
+			r, err := validate(nil, test.p)
+			if test.e != "" {
+				assert.EqualError(t, err, test.e)
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, test.c, r)
+			}
+		})
+	}
+}

extensions/impl/zmq/sink.go

@@ -0,0 +1,88 @@
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build !windows
+
+package zmq
+
+import (
+	"fmt"
+
+	zmq "github.com/pebbe/zmq4"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/errorx"
+)
+
+type zmqSink struct {
+	publisher *zmq.Socket
+	sc        *c
+}
+
+func (m *zmqSink) Provision(ctx api.StreamContext, configs map[string]any) error {
+	sc, err := validate(ctx, configs)
+	if err != nil {
+		return err
+	}
+	m.sc = sc
+	return nil
+}
+
+func (m *zmqSink) Connect(ctx api.StreamContext) (err error) {
+	m.publisher, err = zmq.NewSocket(zmq.PUB)
+	if err != nil {
+		return fmt.Errorf("zmq sink fails to create socket: %v", err)
+	}
+	err = m.publisher.Bind(m.sc.Server)
+	if err != nil {
+		return fmt.Errorf("zmq sink fails to bind to %s: %v", m.sc.Server, err)
+	}
+	ctx.GetLogger().Debugf("zmq sink open")
+	return nil
+}
+
+func (m *zmqSink) Collect(ctx api.StreamContext, item api.RawTuple) error {
+	return m.sendToZmq(ctx, item.Raw())
+}
+
+func (m *zmqSink) sendToZmq(ctx api.StreamContext, v []byte) error {
+	var err error
+	if m.sc.Topic == "" {
+		_, err = m.publisher.SendBytes(v, 0)
+	} else {
+		msgs := [][]byte{
+			[]byte(m.sc.Topic),
+			v,
+		}
+		_, err = m.publisher.SendMessage(msgs)
+	}
+	if err != nil {
+		ctx.GetLogger().Errorf("send to zmq error %v", err)
+		return errorx.NewIOErr(err.Error())
+	}
+	return nil
+}
+
+func (m *zmqSink) Close(_ api.StreamContext) error {
+	if m.publisher != nil {
+		return m.publisher.Close()
+	}
+	return nil
+}
+
+func GetSink() api.Sink {
+	return &zmqSink{}
+}
+
+var _ api.BytesCollector = &zmqSink{}

extensions/impl/zmq/source.go

@@ -0,0 +1,107 @@
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build !windows
+
+package zmq
+
+import (
+	"context"
+	"fmt"
+
+	zmq "github.com/pebbe/zmq4"
+
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/pkg/timex"
+)
+
+type zmqSource struct {
+	subscriber *zmq.Socket
+	sc         *c
+	cancel     context.CancelFunc
+}
+
+func (s *zmqSource) Provision(ctx api.StreamContext, configs map[string]any) error {
+	sc, err := validate(ctx, configs)
+	if err != nil {
+		return err
+	}
+	s.sc = sc
+	return nil
+}
+
+func (s *zmqSource) Connect(ctx api.StreamContext) error {
+	var err error
+	s.subscriber, err = zmq.NewSocket(zmq.SUB)
+	if err != nil {
+		return fmt.Errorf("zmq source fails to create socket: %v", err)
+	}
+	err = s.subscriber.Connect(s.sc.Server)
+	if err != nil {
+		return fmt.Errorf("zmq source fails to connect to %s: %v", s.sc.Server, err)
+	}
+	return nil
+}
+
+func (s *zmqSource) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingestError api.ErrorIngest) error {
+	ctx.GetLogger().Debugf("zmq source subscribe to topic %s", s.sc.Topic)
+	err := s.subscriber.SetSubscribe(s.sc.Topic)
+	if err != nil {
+		return err
+	}
+	for {
+		msgs, err := s.subscriber.RecvMessageBytes(0)
+		if err != nil {
+			id, err := s.subscriber.GetIdentity()
+			ingestError(ctx, fmt.Errorf("zmq source getting message %s error: %v", id, err))
+		} else {
+			rcvTime := timex.GetNow()
+			ctx.GetLogger().Debugf("zmq source receive %v", msgs)
+			var m []byte
+			for i, msg := range msgs {
+				if i == 0 && s.sc.Topic != "" {
+					continue
+				}
+				m = append(m, msg...)
+			}
+			meta := make(map[string]interface{})
+			if s.sc.Topic != "" {
+				meta["topic"] = string(msgs[0])
+			}
+			ingest(ctx, m, meta, rcvTime)
+		}
+		select {
+		case <-ctx.Done():
+			ctx.GetLogger().Infof("zmq source done")
+			if s.subscriber != nil {
+				s.subscriber.Close()
+			}
+		default:
+			// do nothing
+		}
+	}
+}
+
+func (s *zmqSource) Close(_ api.StreamContext) error {
+	if s.cancel != nil {
+		s.cancel()
+	}
+	return nil
+}
+
+func GetSource() api.Source {
+	return &zmqSource{}
+}
+
+var _ api.BytesSource = &zmqSource{}

extensions/sinks/zmq/install.sh

@@ -0,0 +1,64 @@
+#!/bin/sh
+#
+# Copyright 2023-2024 EMQ Technologies Co., Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+set +e -x -u
+
+DISTRO='unknow'
+
+Get_Dist_Name()
+{
+    if grep -Eqii "CentOS" /etc/issue || grep -Eq "CentOS" /etc/*-release; then
+        DISTRO='CentOS'
+    elif grep -Eqi "Red Hat Enterprise Linux Server" /etc/issue || grep -Eq "Red Hat Enterprise Linux Server" /etc/*-release; then
+        DISTRO='RHEL'
+    elif grep -Eqi "Aliyun" /etc/issue || grep -Eq "Aliyun" /etc/*-release; then
+        DISTRO='Aliyun'
+    elif grep -Eqi "Fedora" /etc/issue || grep -Eq "Fedora" /etc/*-release; then
+        DISTRO='Fedora'
+    elif grep -Eqi "Debian" /etc/issue || grep -Eq "Debian" /etc/*-release; then
+        DISTRO='Debian'
+    elif grep -Eqi "Ubuntu" /etc/issue || grep -Eq "Ubuntu" /etc/*-release; then
+        DISTRO='Ubuntu'
+    elif grep -Eqi "Raspbian" /etc/issue || grep -Eq "Raspbian" /etc/*-release; then
+        DISTRO='Raspbian'
+    elif grep -Eqi "Alpine" /etc/issue || grep -Eq "Alpine" /etc/*-release; then
+        DISTRO='Alpine'
+    else
+        DISTRO='unknow'
+    fi
+    echo $DISTRO;
+}
+
+
+Get_Dist_Name
+
+case $DISTRO in \
+    Debian|Ubuntu|Raspbian ) \
+	      apt-get update \
+          && apt-get install -y libczmq-dev 2> /dev/null \
+          && apt-get clean \
+          && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* \
+    ;; \
+    Alpine ) \
+        apk add libzmq \
+    ;; \
+    *) \
+        yum install -y zeromq 2> /dev/null \
+    ;; \
+esac
+
+echo "install success";

extensions/sinks/zmq/zmq.go

@@ -0,0 +1,24 @@
+// Copyright 2021-2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"github.com/lf-edge/ekuiper/contract/v2/api"
+	"github.com/lf-edge/ekuiper/v2/extensions/impl/zmq"
+)
+
+func Zmq() api.Sink {
+	return zmq.GetSink()
+}