refactor: major refactor for the relay defintions.

Signed-off-by: Andrew Neudegg <andrew.neudegg@finbourne.com>

cmd/bridge/subcmd/simulator/bridge.go

@@ -2,25 +2,17 @@
 package simulator
 
 import (
-	"context"
-	"fmt"
-	"math/rand"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/spf13/cobra"
 
 	"github.com/andrewneudegg/delta/cmd/bridge/apputil"
-	"github.com/andrewneudegg/delta/pkg/events"
-	"github.com/andrewneudegg/delta/pkg/queue"
 )
 
 type simulatorOpts struct {
 	Interval time.Duration
 
 	appState *apputil.AppState
-
-	Qimpl queue.Q
 }
 
 // Cmd demonstrates how to configure a new subcommand.
@@ -35,13 +27,6 @@ func Cmd(appState *apputil.AppState) *cobra.Command {
 		Run: func(cmd *cobra.Command, args []string) {
 			appState.Logger.Debug("starting example cmd")
 
-			go simulator(&simulatorOpts{
-				Interval: generationInterval,
-				appState: appState,
-				Qimpl:    queue.NewMemoryQ(),
-			})
-
-			appState.Block(context.TODO())
 		},
 	}
 
@@ -50,32 +35,32 @@ func Cmd(appState *apputil.AppState) *cobra.Command {
 	return cmd
 }
 
-func simulator(opts *simulatorOpts) {
-	r := rand.New(rand.NewSource(0))
+// func simulator(opts *simulatorOpts) {
+// 	r := rand.New(rand.NewSource(0))
 
-	mq := make(chan events.Event)
-	go func() {
-		for {
-			msg := <-mq
-			opts.appState.Logger.Debugf("queuing msg with id: '%s'", msg.GetMessageID())
-			opts.Qimpl.Push(msg)
-		}
-	}()
+// 	mq := make(chan events.Event)
+// 	go func() {
+// 		for {
+// 			msg := <-mq
+// 			opts.appState.Logger.Debugf("queuing msg with id: '%s'", msg.GetMessageID())
+// 			opts.Qimpl.Push(msg)
+// 		}
+// 	}()
 
-	go func() {
-		for {
-			time.Sleep(opts.Interval)
-			opts.appState.Logger.Debug("generating event")
-			msg := &events.EventMsg{
-				ID: uuid.New().String(),
-				Headers: map[string][]string{
-					"User-Agent": {"test"},
-					"Host":       {"test.com"},
-				},
-				URI:     "/test/1/2/3",
-				Content: []byte(fmt.Sprintf("%d", r.Intn(100000))),
-			}
-			mq <- msg
-		}
-	}()
-}
+// 	go func() {
+// 		for {
+// 			time.Sleep(opts.Interval)
+// 			opts.appState.Logger.Debug("generating event")
+// 			msg := &events.EventMsg{
+// 				ID: uuid.New().String(),
+// 				Headers: map[string][]string{
+// 					"User-Agent": {"test"},
+// 					"Host":       {"test.com"},
+// 				},
+// 				URI:     "/test/1/2/3",
+// 				Content: []byte(fmt.Sprintf("%d", r.Intn(100000))),
+// 			}
+// 			mq <- msg
+// 		}
+// 	}()
+// }

pkg/queue/memory.go → pkg/_queue/memory.go

@@ -1,4 +1,4 @@
-package queue
+package _queue
 
 import (
 	"container/list"

pkg/queue/memory_test.go → pkg/_queue/memory_test.go

@@ -1,4 +1,4 @@
-package queue
+package _queue
 
 import (
 	"testing"

pkg/queue/queue.go → pkg/_queue/queue.go

@@ -1,4 +1,4 @@
-package queue
+package _queue
 
 // Q is a any type of queue, memory, RPC, HTTP, Kafka, etc.
 type Q interface {

pkg/relay/httpRelay.go → pkg/_relay/httpRelay.go

@@ -1,4 +1,4 @@
-package relay
+package _relay
 
 import (
 	"context"

pkg/relay/noop.go → pkg/_relay/noop.go

@@ -1,4 +1,4 @@
-package relay
+package _relay
 
 import (
 	"context"

pkg/_relay/relay.go

@@ -0,0 +1,34 @@
+package _relay
+
+import (
+	"context"
+
+	"github.com/andrewneudegg/delta/pkg/events"
+)
+
+// ------------ RELAY -------------
+
+// R is the relay interface, all relays should adhere to this.
+type R interface {
+	Do(context.Context, <-chan events.Event)
+}
+
+// ------------ / RELAY -------------
+
+// Server will serve events to requesting clients.
+type Server interface {
+	events.Distributor
+	Start() error                   // Start the server.
+	Stop(ctx context.Context) error // Stop the server.
+}
+
+// Client provides clients with a mechanism to access events.
+type Client interface {
+	GetNextEvent() (events.Event, error)        // GetNextEvent will return the next available event.
+	GetNextNEvents(int) ([]events.Event, error) // GetNextNEvents will return up to N next events.
+	CompleteEvent(events.Event) error           // CompleteEvent will complete the underlying event if required.
+	CompleteNEvents([]events.Event) error       // CompleteNEvents will complete N underlying events if required.
+	FailEvent(events.Event) error               // FailEvent will inform the RelayServer that the given event could not be processed.
+	FailNEvents([]events.Event) error           // FailNEvents provides a convenience mechanism for bulk failing events.
+	EstimateLoad() (int, error)                 // EstimateLoad provides an estimate for the amount of work being done by this server, between 0 and 100.
+}

pkg/e2e/sink_test/basic_test.go

@@ -7,7 +7,7 @@ import (
 	"time"
 
 	"github.com/andrewneudegg/delta/pkg/events"
-	"github.com/andrewneudegg/delta/pkg/messaging/memory"
+	"github.com/andrewneudegg/delta/pkg/relay/memory"
 	"github.com/andrewneudegg/delta/pkg/sink"
 	"github.com/stretchr/testify/assert"
 )
@@ -27,13 +27,9 @@ func TestE2EMemory(t *testing.T) {
 		}
 	}(distributedMessageChan)
 
-	// phonebook will route all events to the specified channel.
-	phonebook := memory.Phonebook{
-		Ch: distributedMessageChan,
-	}
 	// relay will route messages from incomingEvents to distributedMessages.
 	relay := memory.Relay{}
-	go relay.Do(context.TODO(), incomingEvents, phonebook)
+	go relay.Do(context.TODO(), incomingEvents, distributedMessageChan)
 
 	// sinkServer will catch all events.
 	sinkServer, _ := sink.NewHTTPSinkServer(&sink.HTTPSinkServerConfiguration{
@@ -68,8 +64,6 @@ func TestE2EMemory(t *testing.T) {
 	assert.Equal(t, numEvents, len(distributedMessages))
 }
 
-
-
 // func TestSmoke(t *testing.T) {
 // 	mq := make(chan events.Event)
 // 	re := make(chan events.Event)