databroker: remove unused installation id, close streams when backend is closed (#2062)

databroker/databroker.go

@@ -35,7 +35,6 @@ func (srv *dataBrokerServer) OnConfigChange(cfg *config.Config) {
 func (srv *dataBrokerServer) getOptions(cfg *config.Config) []databroker.ServerOption {
 	cert, _ := cfg.Options.GetDataBrokerCertificate()
 	return []databroker.ServerOption{
-		databroker.WithInstallationID(cfg.Options.InstallationID),
 		databroker.WithSharedKey(cfg.Options.SharedKey),
 		databroker.WithStorageType(cfg.Options.DataBrokerStorageType),
 		databroker.WithStorageConnectionString(cfg.Options.DataBrokerStorageConnectionString),

internal/databroker/config.go

@@ -20,7 +20,6 @@ var (
 )
 
 type serverConfig struct {
-	installationID          string
 	deletePermanentlyAfter  time.Duration
 	secret                  []byte
 	storageType             string
@@ -61,13 +60,6 @@ func WithGetAllPageSize(pageSize int) ServerOption {
 	}
 }
 
-// WithInstallationID sets the installation id in the config.
-func WithInstallationID(installationID string) ServerOption {
-	return func(cfg *serverConfig) {
-		cfg.installationID = installationID
-	}
-}
-
 // WithSharedKey sets the secret in the config.
 func WithSharedKey(sharedKey string) ServerOption {
 	return func(cfg *serverConfig) {

pkg/storage/inmemory/stream.go

@@ -30,6 +30,14 @@ func newRecordStream(ctx context.Context, backend *Backend, version uint64) *rec
 
 		closed: make(chan struct{}),
 	}
+	// if the backend is closed, close the stream
+	go func() {
+		select {
+		case <-stream.closed:
+		case <-backend.closed:
+			_ = stream.Close()
+		}
+	}()
 	return stream
 }
 

pkg/storage/redis/stream.go

@@ -27,7 +27,7 @@ type recordStream struct {
 }
 
 func newRecordStream(ctx context.Context, backend *Backend, version uint64) *recordStream {
-	return &recordStream{
+	stream := &recordStream{
 		ctx:     ctx,
 		backend: backend,
 
@@ -36,6 +36,15 @@ func newRecordStream(ctx context.Context, backend *Backend, version uint64) *rec
 
 		closed: make(chan struct{}),
 	}
+	// if the backend is closed, close the stream
+	go func() {
+		select {
+		case <-stream.closed:
+		case <-backend.closed:
+			_ = stream.Close()
+		}
+	}()
+	return stream
 }
 
 func (stream *recordStream) Close() error {