Cleanup shutdown logic (#56)

Dockerfile

@@ -20,7 +20,6 @@ FROM base as development
 
 RUN ["go", "install", "github.com/githubnemo/CompileDaemon@latest"]
 EXPOSE 9012
-ENV MESSAGING redis
 ENV AUTH ticket
 
 CMD CompileDaemon -log-prefix=false -build="go build ./cmd/icc" -command="./icc"
@@ -37,7 +36,6 @@ LABEL org.opencontainers.image.source="https://github.com/OpenSlides/openslides-
 COPY --from=builder /root/icc .
 COPY --from=builder /root/healthcheck .
 EXPOSE 9007
-ENV MESSAGING redis
 ENV AUTH ticket
 ENTRYPOINT ["/icc"]
 HEALTHCHECK CMD ["/healthcheck"]

README.md

@@ -19,11 +19,10 @@ go build ./cmd/icc
 
 ### With Docker
 
-The docker build uses the redis messaging service, the auth token and the real
-datastore service as default. Either configure it to use the fake services (see
-environment variables below) or make sure the service inside the docker
-container can connect to redis and the datastore-reader. For example with the
-docker argument --network host. The auth-secrets have to given as a file.
+The docker build uses the auth token. Either configure it to use the fake
+services (see environment variables below) or make sure the service inside the
+docker container can connect to redis and the datastore-reader. For example with
+the docker argument --network host. The auth-secrets have to given as a file.
 
 ```
 docker build . --tag openslides-icc
@@ -143,10 +142,6 @@ curl localhost:9007/system/icc/applause/send?meeting_id=1
 
 The argument meeting_id is required.
 
-### Chat 
-
-TODO
-
 
 ## Configuration
 
@@ -167,8 +162,6 @@ The Service uses the following environment variables:
 * `DATASTORE_READER_PORT`: Port of the datastore reader. The default is `9010`.
 * `DATASTORE_READER_PROTOCOL`: Protocol of the datastore reader. The default is
   `http`.
-* `MESSAGING`: Sets the type of messaging service. `fake`(default) or
-  `redis`.
 * `MESSAGE_BUS_HOST`: Host of the redis server. The default is `localhost`.
 * `MESSAGE_BUS_PORT`: Port of the redis server. The default is `6379`.
 * `REDIS_TEST_CONN`: Test the redis connection on startup. Disable on the cloud

go.mod

@@ -3,11 +3,11 @@ module github.com/OpenSlides/openslides-icc-service
 go 1.18
 
 require (
-	github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220413134533-ef848f16f059
+	github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220416070241-9a94afdf8891
 	github.com/gomodule/redigo v1.8.8
 	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/ory/dockertest/v3 v3.8.1
-	github.com/ostcar/topic v0.4.0
+	github.com/ostcar/topic v0.4.1
 	golang.org/x/net v0.0.0-20220412020605-290c469a71a5 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )

go.sum

@@ -7,10 +7,8 @@ github.com/Microsoft/go-winio v0.5.2 h1:a9IhgEQBCUEk6QCdml9CiJGhAws+YwffDHEMp1VM
 github.com/Microsoft/go-winio v0.5.2/go.mod h1:WpS1mjBmmwHBEWmogvA2mj8546UReBk4v8QkMxJ6pZY=
 github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEVMRuU21PR1EtLVZJmdB18Gu3Rw=
 github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
-github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220210150646-5678dc385a7d h1:H73aexBnYtnEw1YLhMu2VkKOGOSBAZUUvdojOSdjqYI=
-github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220210150646-5678dc385a7d/go.mod h1:KDYKuLCbQD/RChC650VcHNItD3WLnzThvodxxN6PZIg=
-github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220413134533-ef848f16f059 h1:rfiRMcsqNB9Qf4XiFtdWQhBh2CpTTwxlDFB27ese+6Y=
-github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220413134533-ef848f16f059/go.mod h1:v8Htxhn7MJgriYTzmajeHeVy//LKn8BlttMMrnZwvdQ=
+github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220416070241-9a94afdf8891 h1:lekthlRTM04PDwlxdSU6f25Ve7QXIWeDxi983uerm9w=
+github.com/OpenSlides/openslides-autoupdate-service v0.4.1-0.20220416070241-9a94afdf8891/go.mod h1:Eo4aEcDlU+avIm3YZ5OrQIJq3H3HcW/mLwId5SPulVI=
 github.com/bits-and-blooms/bitset v1.2.0/go.mod h1:gIdJ4wp64HaoK2YrL1Q5/N7Y16edYb8uY+O0FJTyyDA=
 github.com/cenkalti/backoff/v4 v4.1.2/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/cenkalti/backoff/v4 v4.1.3 h1:cFAlzYUlVYDysBEH2T5hyJZMh3+5+WCBvSnK6Q8UtC4=
@@ -99,8 +97,8 @@ github.com/opencontainers/selinux v1.8.2/go.mod h1:MUIHuUEvKB1wtJjQdOyYRgOnLD2xA
 github.com/opencontainers/selinux v1.10.0/go.mod h1:2i0OySw99QjzBBQByd1Gr9gSjvuho1lHsJxIJ3gGbJI=
 github.com/ory/dockertest/v3 v3.8.1 h1:vU/8d1We4qIad2YM0kOwRVtnyue7ExvacPiw1yDm17g=
 github.com/ory/dockertest/v3 v3.8.1/go.mod h1:wSRQ3wmkz+uSARYMk7kVJFDBGm8x5gSxIhI7NDc+BAQ=
-github.com/ostcar/topic v0.4.0 h1:Ylu6gWGZO7zh9BtIkVHohvZzv0XLGHGHdnhZa4Cp9M4=
-github.com/ostcar/topic v0.4.0/go.mod h1:13aefloBRYAhhb4BWjwb0hMRNx+9QSbdyCJ631ioCW4=
+github.com/ostcar/topic v0.4.1 h1:ORxFOS8BAVKRaeAr3lwYrETQAuKojCUxzWOoBn0CQTw=
+github.com/ostcar/topic v0.4.1/go.mod h1:13aefloBRYAhhb4BWjwb0hMRNx+9QSbdyCJ631ioCW4=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -117,8 +115,8 @@ github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnIn
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
 github.com/syndtr/gocapability v0.0.0-20200815063812-42c35b437635/go.mod h1:hkRG7XYTFWNJGYcbNJQlaLq0fg1yr4J4t/NcTQtrfww=
 github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE=

internal/applause/applause.go

@@ -42,10 +42,10 @@ type Applause struct {
 //
 // The New function is not blocking. The context is used to stop a goroutine
 // that is started by this function.
-func New(b Backend, db datastore.Getter, closed <-chan struct{}) *Applause {
+func New(b Backend, db datastore.Getter) *Applause {
 	notify := Applause{
 		backend:   b,
-		topic:     topic.New(topic.WithClosed[string](closed)),
+		topic:     topic.New[string](),
 		datastore: db,
 	}
 

internal/icchttp/http.go

@@ -72,19 +72,7 @@ func Error(w http.ResponseWriter, err error) {
 }
 
 func isConnectionClose(err error) bool {
-	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
-		// Client closes connection.
-		return true
-	}
-
-	var closing interface {
-		Closing()
-	}
-	if errors.As(err, &closing) {
-		// Server is shutting down.
-		return true
-	}
-	return false
+	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
 }
 
 // AuthMiddleware checks the user id of the request.

internal/notify/notify.go

@@ -40,19 +40,18 @@ type Notify struct {
 //
 // The New function is not blocking. The context is used to stop a goroutine
 // that is started by this function.
-func New(ctx context.Context, b Backend) *Notify {
+func New(b Backend) *Notify {
 	notify := Notify{
 		backend: b,
-		topic:   topic.New(topic.WithClosed[string](ctx.Done())),
+		topic:   topic.New[string](),
 	}
 
-	go notify.listen(ctx)
 	return &notify
 }
 
-// listen waits for Notify messages from the backend and saves them into the
+// Listen waits for Notify messages from the backend and saves them into the
 // topic.
-func (n *Notify) listen(ctx context.Context) {
+func (n *Notify) Listen(ctx context.Context) {
 	for {
 		m, err := n.backend.NotifyReceive(ctx)
 		if err != nil {

internal/notify/notify_test.go

@@ -12,11 +12,12 @@ import (
 )
 
 func TestSend(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
+	shutdownCtx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	backend := newBackendStrub()
-	n := notify.New(ctx, backend)
+	n := notify.New(backend)
+	go n.Listen(shutdownCtx)
 
 	t.Run("invalid json", func(t *testing.T) {
 		defer backend.reset()
@@ -110,11 +111,12 @@ func TestSend(t *testing.T) {
 }
 
 func TestReceive(t *testing.T) {
-	testCtx, cancel := context.WithCancel(context.Background())
+	shutdownCtx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	backend := newBackendStrub()
-	n := notify.New(testCtx, backend)
+	n := notify.New(backend)
+	go n.Listen(shutdownCtx)
 
 	_, next := n.Receive(1, 2)
 

internal/run/run.go

@@ -26,7 +26,13 @@ import (
 func Run(ctx context.Context, environment []string, secret func(name string) (string, error)) error {
 	env := defaultEnv(environment)
 
-	errHandler := buildErrHandler()
+	errHandler := func(err error) {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			return
+		}
+
+		icclog.Info("Error: %v", err)
+	}
 
 	messageBus, err := buildMessageBus(env)
 	if err != nil {
@@ -51,8 +57,10 @@ func Run(ctx context.Context, environment []string, secret func(name string) (st
 
 	backend := redis.New(env["ICC_REDIS_HOST"] + ":" + env["ICC_REDIS_PORT"])
 
-	notifyService := notify.New(ctx, backend)
-	applauseService := applause.New(backend, ds, ctx.Done())
+	notifyService := notify.New(backend)
+	go notifyService.Listen(ctx)
+
+	applauseService := applause.New(backend, ds)
 	go applauseService.Loop(ctx, errHandler)
 	go applauseService.PruneOldData(ctx)
 
@@ -100,7 +108,6 @@ func defaultEnv(environment []string) map[string]string {
 		"DATASTORE_READER_PORT":     "9010",
 		"DATASTORE_READER_PROTOCOL": "http",
 
-		"MESSAGING":        "fake",
 		"MESSAGE_BUS_HOST": "localhost",
 		"MESSAGE_BUS_PORT": "6379",
 		"REDIS_TEST_CONN":  "true",
@@ -145,21 +152,6 @@ func secret(name string, getSecret func(name string) (string, error), dev bool)
 	return s, nil
 }
 
-func buildErrHandler() func(err error) {
-	return func(err error) {
-		if errors.Is(err, context.Canceled) {
-			return
-		}
-
-		var closing interface {
-			Closing()
-		}
-		if !errors.As(err, &closing) {
-			icclog.Info("Error: %v", err)
-		}
-	}
-}
-
 // buildAuth returns the auth service needed by the http server.
 func buildAuth(
 	ctx context.Context,
@@ -193,7 +185,7 @@ func buildAuth(
 
 		icclog.Info("Auth Service: %s", url)
 
-		a, err := auth.New(url, ctx.Done(), []byte(tokenKey), []byte(cookieKey))
+		a, err := auth.New(url, []byte(tokenKey), []byte(cookieKey))
 		if err != nil {
 			return nil, fmt.Errorf("creating auth connection: %w", err)
 		}
@@ -231,26 +223,12 @@ type messageBus interface {
 }
 
 func buildMessageBus(env map[string]string) (messageBus, error) {
-	serviceName := env["MESSAGING"]
-	icclog.Info("Messaging Service: %s", serviceName)
-
-	var conn messageBusRedis.Connection
-	switch serviceName {
-	case "redis":
-		redisAddress := env["MESSAGE_BUS_HOST"] + ":" + env["MESSAGE_BUS_PORT"]
-		c := messageBusRedis.NewConnection(redisAddress)
-		if env["REDIS_TEST_CONN"] == "true" {
-			if err := c.TestConn(); err != nil {
-				return nil, fmt.Errorf("connect to redis: %w", err)
-			}
+	redisAddress := env["MESSAGE_BUS_HOST"] + ":" + env["MESSAGE_BUS_PORT"]
+	conn := messageBusRedis.NewConnection(redisAddress)
+	if env["REDIS_TEST_CONN"] == "true" {
+		if err := conn.TestConn(); err != nil {
+			return nil, fmt.Errorf("connect to redis: %w", err)
 		}
-
-		conn = c
-
-	case "fake":
-		conn = messageBusRedis.BlockingConn{}
-	default:
-		return nil, fmt.Errorf("unknown messagin service `%s`", serviceName)
 	}
 
 	return &messageBusRedis.Redis{Conn: conn}, nil