move writer interface and redis publish from api module as these are for the server side.

app/driver/redis/publisher.go

@@ -0,0 +1,37 @@
+package redis
+
+import (
+	"context"
+
+	"github.com/eosswedenorg/thalos/api"
+	. "github.com/eosswedenorg/thalos/api/redis"
+
+	"github.com/redis/go-redis/v9"
+)
+
+type Publisher struct {
+	pipeline redis.Pipeliner
+	ctx      context.Context
+	ns       Namespace
+}
+
+func NewPublisher(ctx context.Context, client *redis.Client, ns Namespace) *Publisher {
+	return &Publisher{
+		pipeline: client.Pipeline(),
+		ctx:      ctx,
+		ns:       ns,
+	}
+}
+
+func (r *Publisher) Write(channel api.Channel, payload []byte) error {
+	return r.pipeline.Publish(r.ctx, r.ns.NewKey(channel).String(), payload).Err()
+}
+
+func (r *Publisher) Flush() error {
+	_, err := r.pipeline.Exec(r.ctx)
+	return err
+}
+
+func (r *Publisher) Close() error {
+	return r.Flush()
+}

app/driver/redis/publisher_test.go

@@ -0,0 +1,28 @@
+package redis
+
+import (
+	"context"
+	"testing"
+
+	"github.com/eosswedenorg/thalos/api"
+	. "github.com/eosswedenorg/thalos/api/redis"
+
+	"github.com/go-redis/redismock/v9"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPublisher_Write(t *testing.T) {
+	client, mock := redismock.NewClientMock()
+
+	pub := NewPublisher(context.Background(), client, Namespace{ChainID: "id"})
+
+	mock.MatchExpectationsInOrder(true)
+	mock.ExpectPublish("ship::id::test", []byte("some string")).SetVal(0)
+	mock.ExpectPublish("ship::id::test2", []byte("some other string")).SetVal(0)
+
+	assert.NoError(t, pub.Write(api.Channel{"test"}, []byte("some string")))
+	assert.NoError(t, pub.Write(api.Channel{"test2"}, []byte("some other string")))
+	assert.NoError(t, pub.Flush())
+
+	assert.NoError(t, mock.ExpectationsWereMet())
+}

app/driver/writer.go

@@ -0,0 +1,21 @@
+package driver
+
+import "github.com/eosswedenorg/thalos/api"
+
+// Writer interface defines the required methods
+// to send messages over an channel.
+//
+// This is a low-level interface typically implemented by backend drivers
+type Writer interface {
+	// Write writes a message over a channel.
+	// The message may or may not be buffered depending on the implementation.
+	Write(channel api.Channel, payload []byte) error
+
+	// Flush writes any buffered messages to the channel.
+	// If the implementation does not support buffering. this is a noop.
+	Flush() error
+
+	// Close closes the writer
+	// Any blocked Flush or Write operations will be unblocked.
+	Close() error
+}

app/ship_processor.go

@@ -7,6 +7,7 @@ import (
 	"github.com/eosswedenorg/thalos/api"
 	"github.com/eosswedenorg/thalos/api/message"
 	"github.com/eosswedenorg/thalos/app/abi"
+	"github.com/eosswedenorg/thalos/app/driver"
 
 	log "github.com/sirupsen/logrus"
 
@@ -38,7 +39,7 @@ type ShipProcessor struct {
 	abi *abi.AbiManager
 
 	// Writer to send messages to.
-	writer api.Writer
+	writer driver.Writer
 
 	// Encoder used to encode messages
 	encode message.Encoder
@@ -51,7 +52,7 @@ type ShipProcessor struct {
 }
 
 // SpawnProcessor creates a new ShipProccessor that consumes the shipclient.Stream passed to it.
-func SpawnProccessor(shipStream *shipclient.Stream, writer api.Writer, abi *abi.AbiManager, codec message.Codec) *ShipProcessor {
+func SpawnProccessor(shipStream *shipclient.Stream, writer driver.Writer, abi *abi.AbiManager, codec message.Codec) *ShipProcessor {
 	processor := &ShipProcessor{
 		abi:           abi,
 		writer:        writer,