Merge LogStructured in SQLLog (#194)

pkg/kine/broadcaster/broadcaster.go

@@ -5,21 +5,21 @@ import (
 	"sync"
 )
 
-type ConnectFunc func() (chan interface{}, error)
+type ConnectFunc[T any] func(ctx context.Context) (chan T, error)
 
-type Broadcaster struct {
+type Broadcaster[T any] struct {
 	sync.Mutex
 	running bool
-	subs    map[chan interface{}]struct{}
+	subs    map[chan T]struct{}
 }
 
-func (b *Broadcaster) Subscribe(ctx context.Context) (<-chan interface{}, error) {
+func (b *Broadcaster[T]) Subscribe(ctx context.Context) (<-chan T, error) {
 	b.Lock()
 	defer b.Unlock()
 
-	sub := make(chan interface{}, 100)
+	sub := make(chan T, 100)
 	if b.subs == nil {
-		b.subs = map[chan interface{}]struct{}{}
+		b.subs = map[chan T]struct{}{}
 	}
 	b.subs[sub] = struct{}{}
 	context.AfterFunc(ctx, func() {
@@ -31,18 +31,18 @@ func (b *Broadcaster) Subscribe(ctx context.Context) (<-chan interface{}, error)
 	return sub, nil
 }
 
-func (b *Broadcaster) unsub(sub chan interface{}) {
+func (b *Broadcaster[T]) unsub(sub chan T) {
 	if _, ok := b.subs[sub]; ok {
 		close(sub)
 		delete(b.subs, sub)
 	}
 }
 
-func (b *Broadcaster) Start(connect ConnectFunc) error {
+func (b *Broadcaster[T]) Start(ctx context.Context, connect ConnectFunc[T]) error {
 	b.Lock()
 	defer b.Unlock()
 
-	c, err := connect()
+	c, err := connect(ctx)
 	if err != nil {
 		return err
 	}
@@ -52,7 +52,7 @@ func (b *Broadcaster) Start(connect ConnectFunc) error {
 	return nil
 }
 
-func (b *Broadcaster) stream(ch chan interface{}) {
+func (b *Broadcaster[T]) stream(ch chan T) {
 	for item := range ch {
 		b.publish(item)
 	}
@@ -65,7 +65,7 @@ func (b *Broadcaster) stream(ch chan interface{}) {
 	b.running = false
 }
 
-func (b *Broadcaster) publish(item interface{}) {
+func (b *Broadcaster[T]) publish(item T) {
 	b.Lock()
 	defer b.Unlock()
 

pkg/kine/drivers/sqlite/sqlite.go

@@ -10,9 +10,8 @@ import (
 	"time"
 
 	"github.com/canonical/k8s-dqlite/pkg/kine/drivers/generic"
-	"github.com/canonical/k8s-dqlite/pkg/kine/logstructured"
-	"github.com/canonical/k8s-dqlite/pkg/kine/logstructured/sqllog"
 	"github.com/canonical/k8s-dqlite/pkg/kine/server"
+	"github.com/canonical/k8s-dqlite/pkg/kine/sqllog"
 	"github.com/mattn/go-sqlite3"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -98,7 +97,7 @@ func NewVariant(ctx context.Context, driverName, dataSourceName string, connecti
 		}
 	}
 
-	return logstructured.New(sqllog.New(dialect)), dialect, nil
+	return sqllog.New(dialect), dialect, nil
 }
 
 // setup performs table setup, which may include creation of the Kine table if

pkg/kine/endpoint/endpoint.go

@@ -76,11 +76,10 @@ func Listen(ctx context.Context, config Config) (ETCDConfig, error) {
 
 	go func() {
 		if err := grpcServer.Serve(listener); err != nil {
-			logrus.Errorf("Kine server shutdown: %v", err)
+			logrus.Errorf("unexpected server shutdown: %v", err)
 		}
-		listener.Close()
-		grpcServer.Stop()
 	}()
+	context.AfterFunc(ctx, grpcServer.Stop)
 
 	return ETCDConfig{
 		LeaderElect: leaderelect,
@@ -145,8 +144,8 @@ func ListenAndReturnBackend(ctx context.Context, config Config) (ETCDConfig, ser
 			logrus.Errorf("Kine server shutdown: %v", err)
 		}
 		listener.Close()
-		grpcServer.Stop()
 	}()
+	context.AfterFunc(ctx, grpcServer.Stop)
 
 	return ETCDConfig{
 		LeaderElect: leaderelect,