Lightweight event framework for microservices

.github/workflows/go.yml

@@ -18,7 +18,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: 1.19
+          go-version: '1.20'
 
       - name: Build and Test
         run: make

Makefile

@@ -21,7 +21,7 @@ lint: ${LINT}
 ${MOCKERY}:
 		$(VGO) install github.com/vektra/mockery/v2@latest
 ${LINT}:
-		$(VGO) install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.47.0
+		$(VGO) install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.55.0
 dbmigrate:
 		$(eval DBMIGRATE_PATH := $(shell $(VGO) list -f '{{.Dir}}' github.com/golang-migrate/migrate/v4/database))
 
@@ -35,6 +35,9 @@ endef
 $(eval $(call makemock, $$(DBMIGRATE_PATH),        Driver,             dbmigratemocks))
 $(eval $(call makemock, pkg/httpserver,            GoHTTPServer,       httpservermocks))
 $(eval $(call makemock, pkg/auth,                  Plugin,             authmocks))
+$(eval $(call makemock, pkg/wsserver,              WebSocketChannels,  wsservermocks))
+$(eval $(call makemock, pkg/wsserver,              WebSocketServer,    wsservermocks))
+$(eval $(call makemock, pkg/dbsql,                 CRUD,               crudmocks))
 
 firefly-common: ${GOFILES}
 		$(VGO) build ./pkg/*

examples/ffpubsub.go

@@ -0,0 +1,180 @@
+// Copyright © 2023 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// 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 (
+	"bufio"
+	"context"
+	"database/sql/driver"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"strconv"
+	"sync"
+
+	"github.com/hyperledger/firefly-common/pkg/config"
+	"github.com/hyperledger/firefly-common/pkg/dbsql"
+	"github.com/hyperledger/firefly-common/pkg/eventstreams"
+	"github.com/hyperledger/firefly-common/pkg/fftypes"
+	"github.com/hyperledger/firefly-common/pkg/log"
+	"github.com/hyperledger/firefly-common/pkg/wsserver"
+)
+
+// Run from root of project with:
+// go run ./examples/ffpubsub.go
+func main() {
+	ctx := context.Background()
+
+	// Set ourselves up using an im-memory DB
+	mgr, ims, done := setup(ctx)
+	defer done()
+
+	// This demo:  Just create a single event stream
+	// Real world: You would create ffapi.Route objects for the CRUD actions on mgr
+	_, err := mgr.UpsertStream(ctx, &eventstreams.EventStreamSpec[pubSubConfig]{
+		Name: ptrTo("demo"),
+	})
+	assertNoError(err)
+
+	// Read lines from standard in, and pass them to all active routines
+	log.L(ctx).Infof("Waiting for lines to publish on stdin...")
+	scanner := bufio.NewScanner(os.Stdin)
+	for scanner.Scan() {
+		ims.newMessages.L.Lock()
+		index := len(ims.messages)
+		ims.messages = append(ims.messages, scanner.Text())
+		ims.newMessages.Broadcast()
+		ims.newMessages.L.Unlock()
+		log.L(ctx).Infof("Published %.12d", index)
+	}
+}
+
+type pubSubESManager = eventstreams.Manager[pubSubConfig]
+
+type pubSubConfig struct {
+	// no extra config
+}
+
+func (psc *pubSubConfig) Scan(src interface{}) error {
+	return fftypes.JSONScan(src, psc)
+}
+
+func (psc *pubSubConfig) Value() (driver.Value, error) {
+	return fftypes.JSONValue(psc)
+}
+
+type pubSubMessage struct {
+	Message string `json:"message"`
+}
+
+// The most trivial data store
+type inMemoryStream struct {
+	messages    []string
+	newMessages sync.Cond
+}
+
+func (ims *inMemoryStream) Validate(_ context.Context, _ *pubSubConfig) error {
+	return nil // no config defined in pubSubConfig to validate
+}
+
+func (ims *inMemoryStream) Run(_ context.Context, _ *eventstreams.EventStreamSpec[pubSubConfig], checkpointSequenceID string, deliver eventstreams.Deliver[pubSubMessage]) (err error) {
+	var index int
+	if checkpointSequenceID != "" {
+		index, err = strconv.Atoi(checkpointSequenceID)
+		if err != nil {
+			return fmt.Errorf("bad checkpoint '%s': %s", checkpointSequenceID, err)
+		}
+	}
+	for {
+		// Get the next message
+		var nextMsg string
+		ims.newMessages.L.Lock()
+		for nextMsg == "" {
+			ims.newMessages.Wait()
+			if index <= len(ims.messages)-1 {
+				nextMsg = ims.messages[index]
+				index++
+			}
+		}
+		ims.newMessages.L.Unlock()
+
+		// Deliver it
+		if deliver([]*eventstreams.Event[pubSubMessage]{
+			{
+				EventCommon: eventstreams.EventCommon{
+					Topic:      "topic1",
+					SequenceID: fmt.Sprintf("%.12d", index),
+				},
+				Data: &pubSubMessage{
+					Message: nextMsg,
+				},
+			},
+		}) == eventstreams.Exit {
+			return nil
+		}
+	}
+}
+
+func ptrTo[T any](v T) *T {
+	return &v
+}
+
+func setup(ctx context.Context) (pubSubESManager, *inMemoryStream, func()) {
+	// Use SQLite in-memory DB
+	conf := config.RootSection("ffpubsub")
+	eventstreams.InitConfig(conf)
+	dbConf := conf.SubSection("sqlite")
+	dbsql.InitSQLiteConfig(dbConf)
+	dbConf.Set(dbsql.SQLConfMigrationsAuto, true)
+	dbConf.Set(dbsql.SQLConfDatasourceURL, "file::memory:")
+	dbConf.Set(dbsql.SQLConfMigrationsAuto, true)
+	dbConf.Set(dbsql.SQLConfMigrationsDirectory, "./test/es_demo_migrations")
+	dbConf.Set(dbsql.SQLConfMaxConnections, 1)
+
+	sql, err := dbsql.NewSQLiteProvider(ctx, dbConf)
+	assertNoError(err)
+
+	wsServer := wsserver.NewWebSocketServer(ctx)
+	server := httptest.NewServer(http.HandlerFunc(wsServer.Handler))
+	u, err := url.Parse(server.URL)
+	assertNoError(err)
+	u.Scheme = "ws"
+	log.L(ctx).Infof("Running on: %s", u)
+
+	p := eventstreams.NewEventStreamPersistence[pubSubConfig](sql)
+	c := eventstreams.GenerateConfig(ctx)
+	ims := &inMemoryStream{
+		messages:    []string{},
+		newMessages: *sync.NewCond(new(sync.Mutex)),
+	}
+	mgr, err := eventstreams.NewEventStreamManager[pubSubConfig, pubSubMessage](ctx, c, p, wsServer, ims)
+	assertNoError(err)
+	return mgr, ims, func() {
+		log.L(ctx).Infof("Shutting down")
+		server.Close()
+		p.Close()
+	}
+}
+
+// trivial error handling in this example
+func assertNoError(err error) {
+	if err != nil {
+		panic(err)
+	}
+}

go.mod

@@ -1,36 +1,38 @@
 module github.com/hyperledger/firefly-common
 
-go 1.19
+go 1.20
 
 require (
 	github.com/DATA-DOG/go-sqlmock v1.5.0
-	github.com/Masterminds/squirrel v1.5.3
+	github.com/Masterminds/squirrel v1.5.4
 	github.com/aidarkhanov/nanoid v1.0.8
 	github.com/docker/go-units v0.5.0
 	github.com/fsnotify/fsnotify v1.6.0
 	github.com/getkin/kin-openapi v0.116.0
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-resty/resty/v2 v2.7.0
-	github.com/golang-migrate/migrate/v4 v4.15.2
+	github.com/golang-migrate/migrate/v4 v4.16.1
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/mux v1.8.0
 	github.com/gorilla/websocket v1.5.0
+	github.com/hyperledger/firefly-transaction-manager v1.3.4
 	github.com/jarcoal/httpmock v1.2.0
 	github.com/karlseguin/ccache v2.0.3+incompatible
-	github.com/mattn/go-sqlite3 v1.14.10
+	github.com/mattn/go-sqlite3 v1.14.16
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.11.1
+	github.com/prometheus/client_golang v1.13.0
 	github.com/rs/cors v1.8.2
 	github.com/santhosh-tekuri/jsonschema/v5 v5.0.2
-	github.com/sirupsen/logrus v1.9.0
+	github.com/sirupsen/logrus v1.9.2
 	github.com/spf13/cobra v1.6.1
 	github.com/spf13/viper v1.14.0
 	github.com/stretchr/testify v1.8.1
 	github.com/x-cray/logrus-prefixed-formatter v0.5.2
-	gitlab.com/hfuss/mux-prometheus v0.0.4
-	golang.org/x/crypto v0.1.0
-	golang.org/x/text v0.8.0
+	gitlab.com/hfuss/mux-prometheus v0.0.5
+	golang.org/x/crypto v0.7.0
+	golang.org/x/text v0.9.0
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
+	gotest.tools v2.2.0+incompatible
 )
 
 require (
@@ -40,6 +42,7 @@ require (
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/swag v0.22.3 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
@@ -57,25 +60,26 @@ require (
 	github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect
+	github.com/oklog/ulid/v2 v2.1.0 // indirect
 	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/onsi/gomega v1.19.0 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.5 // indirect
 	github.com/perimeterx/marshmallow v1.1.4 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/common v0.30.0 // indirect
-	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/prometheus/common v0.37.0 // indirect
+	github.com/prometheus/procfs v0.8.0 // indirect
 	github.com/spf13/afero v1.9.2 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/subosito/gotenv v1.4.1 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
-	golang.org/x/net v0.8.0 // indirect
-	golang.org/x/sys v0.6.0 // indirect
-	golang.org/x/term v0.6.0 // indirect
+	golang.org/x/net v0.10.0 // indirect
+	golang.org/x/sys v0.8.0 // indirect
+	golang.org/x/term v0.8.0 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect