Add a client for sending internal events

.github/workflows/build-publish.yml

@@ -16,14 +16,8 @@ jobs:
       uses: actions/setup-python@v4
       with:
         python-version: "3.9"
-    - name: Install pypa/build
-      run: >-
-        python3 -m
-        pip install
-        build
-        --user
     - name: Build a binary wheel and a source tarball
-      run: python3 -m build
+      run: make build
     - name: Store the distribution packages
       uses: actions/upload-artifact@v4
       with:

.github/workflows/main.yml

@@ -27,3 +27,19 @@ jobs:
         run: pip install -U setuptools
       - name: Run Tox
         run: tox
+  check-golang:
+    # The type of runner that the job will run on
+    runs-on: ubuntu-latest
+    # Steps represent a sequence of tasks that will be executed as part of the job
+    steps:
+      - uses: actions/checkout@v2
+      - name: Lint Go code
+        run: make check
+  run-build:
+    # The type of runner that the job will run on
+    runs-on: ubuntu-latest
+    # Steps represent a sequence of tasks that will be executed as part of the job
+    steps:
+      - uses: actions/checkout@v2
+      - name: Build python module with Go code
+        run: make build

.gitignore

@@ -14,6 +14,7 @@ __pycache__/*
 */.ipynb_checkpoints/*
 .DS_Store
 .tags
+bin/
 
 # Project files
 .ropeproject

MANIFEST.in

@@ -0,0 +1,7 @@
+prune build
+prune docs/_build
+prune docs/api
+recursive-include *.go
+include go.mod go.sum
+graft src/etos_lib/bindings/
+global-exclude *.pyc *.o

Makefile

@@ -0,0 +1,65 @@
+export GOBIN := $(CURDIR)/bin
+export PYBUILD := $(CURDIR)/bin
+export VENVDIR := $(CURDIR)/.buildenv
+export BINDINGS := $(CURDIR)/src/etos_lib/bindings
+export BUILD_MESSAGING := go build --buildmode=c-shared -o $(BINDINGS)/messaging/client.so ./cmd/messaging/main.go
+
+BUILD_PYTHON = -m build
+VIRTUALENV = $(VENVDIR)/bin/python
+GOLANGCI_LINT = $(GOBIN)/golangci-lint
+GOLANGCI_LINT_VERSION = v1.64.6
+
+.PHONY: all
+all: check build
+
+.PHONY: check
+check: staticcheck test
+
+.PHONY: test
+test:
+	go test -cover -timeout 30s -race $(shell go list ./... | grep -v "etos-library/test")
+
+.PHONY: tidy
+tidy:
+	go mod tidy
+
+.PHONY: check-dirty
+check-dirty:
+	$(GIT) diff --exit-code HEAD
+
+.PHONY: staticcheck
+staticcheck: $(GOLANGCI_LINT)
+	$(GOLANGCI_LINT) run
+
+.PHONY: clean
+clean:
+	$(RM) $(GOBIN)/*
+	$(RM) -r $(BINDINGS)
+	$(RM) -r $(VENVDIR)
+	$(RM) -r dist
+
+.PHONY: build
+build: build-bindings build-python
+
+.PHONY: build-bindings
+build-bindings:
+	$(BUILD_MESSAGING)
+
+.PHONY: build-python
+build-python: $(BUILD_PYTHON)
+	$(VIRTUALENV) $(BUILD_PYTHON)
+
+$(GOLANGCI_LINT):
+	mkdir -p $(dir $@)
+	curl -sfL \
+			https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh \
+		| sh -s -- -b $(GOBIN) $(GOLANGCI_LINT_VERSION)
+
+$(BUILD_PYTHON): $(VIRTUALENV)
+	mkdir -p $(dir $@)
+	$(VIRTUALENV) -m pip install build
+
+$(VIRTUALENV):
+	mkdir -p $(dir $@)
+	pip install virtualenv
+	python -m virtualenv $(VENVDIR)

cmd/messaging/main.go

@@ -0,0 +1,27 @@
+// Copyright Axis Communications AB.
+//
+// For a full list of individual contributors, please see the commit history.
+//
+// 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 (
+	"C"
+
+	// This import will make sure that we build the bindings on make build.
+	_ "github.com/eiffel-community/etos-library/pkg/bindings"
+)
+
+// main is required for cgo to work.
+func main() {
+}

go.mod

@@ -0,0 +1,21 @@
+module github.com/eiffel-community/etos-library
+
+go 1.22.1
+
+require (
+	github.com/rabbitmq/rabbitmq-stream-go-client v1.5.0
+	github.com/sirupsen/logrus v1.9.3
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
+	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	github.com/stretchr/testify v1.9.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+)

go.sum

@@ -0,0 +1,63 @@
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
+github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
+github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
+github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
+github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad h1:a6HEuzUHeKH6hwfN/ZoQgRgVIWFJljSWa/zetS2WTvg=
+github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/onsi/ginkgo/v2 v2.22.2 h1:/3X8Panh8/WwhU/3Ssa6rCKqPLuAkVY2I0RoyDLySlU=
+github.com/onsi/ginkgo/v2 v2.22.2/go.mod h1:oeMosUL+8LtarXBHu/c0bx2D/K9zyQ6uX3cTyztHwsk=
+github.com/onsi/gomega v1.36.2 h1:koNYke6TVk6ZmnyHrCXba/T/MoLBXFjeC1PtvYgw0A8=
+github.com/onsi/gomega v1.36.2/go.mod h1:DdwyADRjrc825LhMEkD76cHR5+pUnjhUN8GlHlRPHzY=
+github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM=
+github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rabbitmq/rabbitmq-stream-go-client v1.5.0 h1:2UWryxhtQmYA3Bx2iajQCre3yQbARiSikpC/8iWbu3k=
+github.com/rabbitmq/rabbitmq-stream-go-client v1.5.0/go.mod h1:KDXSNVSqj4QNg6TNMBnQQ/oWHaxLjUI1520j68SyEcY=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
+github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
+github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
+golang.org/x/tools v0.28.0 h1:WuB6qZ4RPCQo5aP3WdKZS7i595EdWqWR8vqJTlwTVK8=
+golang.org/x/tools v0.28.0/go.mod h1:dcIOrVd3mfQKTgrDVQHqCPMWy6lnhfhtX3hLXYVLfRw=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/bindings/publisher.go

@@ -0,0 +1,135 @@
+// Copyright Axis Communications AB.
+//
+// For a full list of individual contributors, please see the commit history.
+//
+// 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 bindings
+
+/*
+#include <stdlib.h>
+#include <stdint.h>
+*/
+import "C"
+
+import (
+	"log/slog"
+	"os"
+	"runtime/cgo"
+
+	"github.com/eiffel-community/etos-library/pkg/stream"
+)
+
+// StreamHandler is a struct that holds the logger, connection string, stream name, streamer,
+// publisher, and consumer.
+// It is used to create a new publisher and publish messages to a stream from python.
+// For using the streamers in Go code please refer to pkg/stream/stream.go.
+//
+// For usage examples view the python code in src/etos_lib/messaging/publisher.py.
+type StreamHandler struct {
+	logger           *slog.Logger
+	connectionString string
+	streamName       string
+	streamer         stream.Streamer
+	publisher        stream.Publisher
+	consumer         stream.Consumer
+}
+
+// New creates a new publisher and returns a pointer to it.
+//
+//export New
+func New(connectionString, streamName *C.char) (C.uintptr_t, bool) {
+	logger := newLogger()
+	streamer, err := stream.NewRabbitMQStreamer(C.GoString(streamName), C.GoString(connectionString), logger)
+	if err != nil {
+		return C.uintptr_t(0), false
+	}
+	return C.uintptr_t(cgo.NewHandle(&StreamHandler{
+		logger:   logger,
+		streamer: streamer,
+	})), true
+}
+
+// Publisher creates a new publisher and starts it.
+//
+//export Publisher
+func Publisher(p C.uintptr_t, name *C.char) bool {
+	h := cgo.Handle(p)
+	handler, ok := h.Value().(*StreamHandler)
+	if !ok {
+		// Creating a new logger here since if handler is nil, we don't have access to one.
+		newLogger().Error("Failed to get stream handler from memory")
+		return false
+	}
+	publisher, err := handler.streamer.Publisher(C.GoString(name))
+	if err != nil {
+		handler.logger.Error("Failed to create publisher", slog.Any("Error", err))
+		return false
+	}
+	handler.publisher = publisher
+	if err = publisher.Start(); err != nil {
+		handler.logger.Error("Failed to start publisher", slog.Any("Error", err))
+		return false
+	}
+	return true
+}
+
+// Publish a message to the stream.
+//
+//export Publish
+func Publish(p C.uintptr_t, event, identifier, eventType, meta *C.char) bool {
+	h := cgo.Handle(p)
+	handler, ok := h.Value().(*StreamHandler)
+	if !ok {
+		// Creating a new logger here since if handler is nil, we don't have access to one.
+		newLogger().Error("Failed to get stream handler from memory")
+		return false
+	}
+	message := C.GoString(event)
+	filter := stream.Filter{
+		Identifier: C.GoString(identifier),
+		Type:       C.GoString(eventType),
+		Meta:       C.GoString(meta),
+	}
+	handler.publisher.Publish([]byte(message), filter)
+	return true
+}
+
+// Close closes the publisher and the streamer.
+//
+//export Close
+func Close(p C.uintptr_t) bool {
+	h := cgo.Handle(p)
+	// Delete the handle here, since it shall not be reused after close.
+	defer h.Delete()
+	handler, ok := h.Value().(*StreamHandler)
+	if !ok {
+		// Creating a new logger here since if handler is nil, we don't have access to one.
+		newLogger().Error("Failed to get stream handler from memory")
+		return false
+	}
+	if handler.publisher != nil {
+		handler.publisher.Close()
+	}
+	if handler.consumer != nil {
+		handler.consumer.Close()
+	}
+	if handler.streamer != nil {
+		handler.streamer.Close()
+	}
+	return true
+}
+
+// newLogger creates a new logger.
+func newLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(os.Stdout, nil))
+}