Events

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/Masterminds/sprig v2.22.0+incompatible
 	github.com/ReneKroon/ttlcache/v2 v2.11.0
 	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d // indirect
+	github.com/asim/go-micro/plugins/events/nats/v4 v4.0.0-20220118152736-9e0be6c85d75 // indirect
 	github.com/aws/aws-sdk-go v1.42.39
 	github.com/beevik/etree v1.1.0
 	github.com/bluele/gcache v0.0.2
@@ -48,6 +49,7 @@ require (
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/go-testing-interface v1.14.1 // indirect
 	github.com/mitchellh/mapstructure v1.4.3
+	github.com/nats-io/nats-streaming-server v0.24.1 // indirect
 	github.com/onsi/ginkgo v1.16.5
 	github.com/onsi/gomega v1.18.0
 	github.com/pkg/errors v0.9.1
@@ -65,17 +67,18 @@ require (
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/tus/tusd v1.8.0
 	github.com/wk8/go-ordered-map v0.2.0
+	go-micro.dev/v4 v4.3.1-0.20211108085239-0c2041e43908 // indirect
 	go.mongodb.org/mongo-driver v1.7.2 // indirect
 	go.opencensus.io v0.23.0
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.28.0
 	go.opentelemetry.io/otel v1.3.0
 	go.opentelemetry.io/otel/exporters/jaeger v1.3.0
 	go.opentelemetry.io/otel/sdk v1.3.0
 	go.opentelemetry.io/otel/trace v1.3.0
-	golang.org/x/crypto v0.0.0-20211215165025-cf75a172585e
+	golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce
 	golang.org/x/oauth2 v0.0.0-20211005180243-6b3c2da341f1
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e
+	golang.org/x/sys v0.0.0-20220114195835-da31bd327af9
 	golang.org/x/term v0.0.0-20210916214954-140adaaadfaf
 	google.golang.org/genproto v0.0.0-20211021150943-2b146023228c
 	google.golang.org/grpc v1.42.0

internal/grpc/interceptors/eventsmiddleware/conversion.go

@@ -0,0 +1,16 @@
+package eventsmiddleware
+
+import (
+	collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
+	"github.com/cs3org/reva/pkg/events"
+)
+
+// ShareCreated converts response to event
+func ShareCreated(r *collaboration.CreateShareResponse) events.ShareCreated {
+	return events.ShareCreated{
+		Sharer: r.Share.Creator,
+		//Sharee: r.Share.Grantee, // TODO: unmarshaling fails -> find out why
+		ItemID: r.Share.ResourceId,
+		CTime:  r.Share.Ctime,
+	}
+}

internal/grpc/interceptors/eventsmiddleware/events.go

@@ -0,0 +1,62 @@
+package eventsmiddleware
+
+import (
+	"context"
+
+	"go-micro.dev/v4/util/log"
+	"google.golang.org/grpc"
+
+	collaboration "github.com/cs3org/go-cs3apis/cs3/sharing/collaboration/v1beta1"
+	"github.com/cs3org/reva/pkg/events"
+	"github.com/cs3org/reva/pkg/events/server"
+	"github.com/cs3org/reva/pkg/rgrpc"
+)
+
+const (
+	defaultPriority = 200
+)
+
+func init() {
+	rgrpc.RegisterUnaryInterceptor("eventsmiddleware", NewUnary)
+}
+
+// NewUnary returns a new unary interceptor that emits events when needed
+func NewUnary(m map[string]interface{}) (grpc.UnaryServerInterceptor, int, error) {
+	stream, err := server.NewNatsStream()
+	if err != nil {
+		return nil, 0, err
+	}
+
+	interceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+		res, err := handler(ctx, req)
+		if err != nil {
+			return res, err
+		}
+
+		var ev interface{}
+		switch v := res.(type) {
+		case *collaboration.CreateShareResponse:
+			ev = ShareCreated(v)
+		}
+
+		if ev != nil {
+			if err := events.Publish(ev, stream); err != nil {
+				// TODO: should we error here? log? panic?
+				log.Error(err)
+			}
+		}
+
+		return res, nil
+	}
+	return interceptor, defaultPriority, nil
+}
+
+// NewStream returns a new server stream interceptor
+// that creates the application context.
+func NewStream() grpc.StreamServerInterceptor {
+	interceptor := func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
+		// TODO: How to create events when streaming?
+		return handler(srv, ss)
+	}
+	return interceptor
+}

internal/grpc/interceptors/loader/loader.go

@@ -20,6 +20,7 @@ package loader
 
 import (
 	// Load core GRPC services
+	_ "github.com/cs3org/reva/internal/grpc/interceptors/eventsmiddleware"
 	_ "github.com/cs3org/reva/internal/grpc/interceptors/readonly"
 	// Add your own service here
 )

pkg/events/events.go

@@ -0,0 +1,84 @@
+package events
+
+import (
+	"log"
+	"reflect"
+
+	"go-micro.dev/v4/events"
+)
+
+var (
+	// MainQueueName is the name of the main queue
+	// All events will go through here as they are forwarded to the consumer via the
+	// group name
+	// TODO: "fan-out" so not all events go through the same queue? requires investigation
+	MainQueueName = "main-queue"
+
+	// MetadatakeyEventType is the key used for the eventtype in the metadata map of the event
+	MetadatakeyEventType = "eventtype"
+)
+
+// Unmarshaller is the interface events need to fullfill
+type Unmarshaller interface {
+	Unmarshal([]byte) (interface{}, error)
+}
+
+// helper so we don't need to reflect too much on runtime
+type eventInfo struct {
+	Type  reflect.Type
+	Value reflect.Value
+	Raw   interface{}
+}
+
+func (e *eventInfo) New() interface{} {
+	r := reflect.New(reflect.PtrTo(e.Type)).Elem()
+	r.Set(e.Value)
+	return r.Interface()
+}
+
+// Consume returns a channel that will get all events that match the given evs
+// group defines the service type: One group will get exactly one copy of a event that is emitted
+// NOTE: uses reflect on initialization
+func Consume(s events.Stream, group string, evs ...Unmarshaller) (<-chan interface{}, error) {
+	c, err := s.Consume(MainQueueName, events.WithGroup(group))
+	if err != nil {
+		return nil, err
+	}
+
+	registeredEvents := map[string]Unmarshaller{}
+	for _, e := range evs {
+		typ := reflect.TypeOf(e)
+		registeredEvents[typ.String()] = e
+	}
+
+	outchan := make(chan interface{})
+	go func() {
+		for {
+			e := <-c
+			et := e.Metadata[MetadatakeyEventType]
+			ev, ok := registeredEvents[et]
+			if !ok {
+				log.Printf("not registered: %s", et)
+				continue
+			}
+
+			event, err := ev.Unmarshal(e.Payload)
+			if err != nil {
+				log.Printf("can't unmarshal event %v", err)
+				continue
+			}
+
+			outchan <- event
+		}
+	}()
+	return outchan, nil
+}
+
+// Publish publishes the ev to the MainQueue from where it is distributed to all subscribers
+// NOTE: needs to use reflect on runtime
+func Publish(ev interface{}, s events.Stream) error {
+	evName := reflect.TypeOf(ev).String()
+	return s.Publish(MainQueueName, ev, events.WithMetadata(map[string]string{
+		MetadatakeyEventType: evName,
+	}))
+}

pkg/events/example/example.go

@@ -0,0 +1,99 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"github.com/cs3org/reva/pkg/events"
+	"github.com/cs3org/reva/pkg/events/server"
+	microevents "go-micro.dev/v4/events"
+)
+
+func main() {
+	// start server
+	err := server.RunNatsServer()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// get client
+	s, err := server.NewNatsStream()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// needed for syncing
+	wg := &sync.WaitGroup{}
+
+	// consumer a has two instances - it is supposed to get only one event
+	wg.Add(1)
+	go Consumer(s, "a", wg)
+	wg.Add(1)
+	go Consumer(s, "a", wg)
+
+	// consumers b and c are supposed to get the event also
+	wg.Add(1)
+	go Consumer(s, "b", wg)
+	wg.Add(1)
+	go Consumer(s, "c", wg)
+
+	// wait for consumer registration
+	wg.Wait()
+
+	// publish an event
+	sc := events.ShareCreated{
+		//Sharer: "userA",
+		//Sharee:   "userB",
+		ItemID: &provider.ResourceId{
+			StorageId: "storageA",
+			OpaqueId:  "opaqueB",
+		},
+	}
+	if err := events.Publish(sc, s); err != nil {
+		log.Fatal(err)
+	}
+
+	// publish another event
+	sc = events.ShareCreated{
+		//SharerID: "user34",
+		//Sharee:   "user12732",
+		ItemID: &provider.ResourceId{
+			StorageId: "storage44",
+			OpaqueId:  "opaque231",
+		},
+	}
+	if err := events.Publish(sc, s); err != nil {
+		log.Fatal(err)
+	}
+
+	// wait for consumers to log
+	time.Sleep(3 * time.Second)
+
+}
+
+// Consumer consumes from queue
+func Consumer(s microevents.Stream, group string, wg *sync.WaitGroup) {
+	c, err := events.Consume(s, group, events.ShareCreated{})
+	if err != nil {
+		wg.Done()
+		fmt.Println("consumer", group, "can't consume", err)
+		return
+	}
+
+	wg.Done()
+	fmt.Printf("%s) consumer waiting\n", group)
+	for {
+		b := <-c
+
+		// proposed usage
+		switch v := b.(type) {
+		case events.ShareCreated:
+			fmt.Printf("%s) Share created: %+v\n", group, v)
+		default:
+			fmt.Printf("%s) Unregistered event: %T, %+v\n", group, v, v)
+		}
+	}
+}

pkg/events/example/serverconsumer/servercosumer.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/cs3org/reva/pkg/events"
+	"github.com/cs3org/reva/pkg/events/server"
+)
+
+// starts a server and listens for all events
+func main() {
+	// start server
+	err := server.RunNatsServer()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// get client
+	s, err := server.NewNatsStream()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	group := ""
+	c, err := events.Consume(s, group, events.ShareCreated{})
+	if err != nil {
+		fmt.Println("consumer", group, "can't consume", err)
+		return
+	}
+
+	fmt.Println("consumer waiting")
+	for {
+		b := <-c
+
+		// proposed usage
+		switch v := b.(type) {
+		case events.ShareCreated:
+			fmt.Printf("%s) Share created: %+v\n", group, v)
+		default:
+			fmt.Printf("%s) Unregistered event: %+v\n", group, v)
+		}
+	}
+}

pkg/events/server/nats.go

@@ -0,0 +1,20 @@
+package server
+
+import (
+	"github.com/asim/go-micro/plugins/events/nats/v4"
+	"github.com/nats-io/nats-streaming-server/server"
+	"go-micro.dev/v4/events"
+)
+
+// RunNatsServer runs the nats streaming server
+func RunNatsServer() error {
+	// TODO: configurable options
+	_, err := server.RunServerWithOpts(nil, nil)
+	return err
+}
+
+// NewNatsStream returns a streaming client used by `Consume` and `Publish` methods
+func NewNatsStream() (events.Stream, error) {
+	// TODO: configurable options
+	return nats.NewStream(nats.Address("127.0.0.1:4222"), nats.ClusterID("test-cluster"))
+}