Implement shutdown use case

Author: exanubes

client.go

@@ -2,10 +2,7 @@ package appsync
 
 import (
 	"context"
-	"errors"
 
-	"github.com/exanubes/appsync/internal/app/engine"
-	"github.com/exanubes/appsync/internal/app/services/connection"
 	"github.com/exanubes/appsync/internal/app/usecases/publish"
 	"github.com/exanubes/appsync/internal/app/usecases/subscribe"
 	"github.com/exanubes/appsync/internal/composition"
@@ -17,9 +14,7 @@ const (
 )
 
 type appsync_client struct {
-	transport connection.Connection
-	runtime   *engine.Engine
-	usecases  *composition.UseCases
+	usecases *composition.UseCases
 }
 
 func (client *appsync_client) Publish(ctx context.Context, input PublishCommandInput) error {
@@ -50,5 +45,5 @@ func (client *appsync_client) Subscribe(ctx context.Context, input SubscribeComm
 }
 
 func (client *appsync_client) Close(ctx context.Context) error {
-	return errors.Join(client.runtime.Close(ctx), client.transport.Close())
+	return client.usecases.Shutdown.Execute(ctx)
 }

connect.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"net/url"
 
+	"github.com/exanubes/appsync/authorizer"
 	"github.com/exanubes/appsync/internal/app"
 	"github.com/exanubes/appsync/internal/app/engine"
 	"github.com/exanubes/appsync/internal/app/heartbeat"
@@ -14,8 +15,8 @@ import (
 	"github.com/exanubes/appsync/internal/app/runtime"
 	"github.com/exanubes/appsync/internal/app/services/connection"
 	"github.com/exanubes/appsync/internal/app/services/io"
+	"github.com/exanubes/appsync/internal/app/usecases/shutdown"
 	"github.com/exanubes/appsync/internal/composition"
-	"github.com/exanubes/appsync/authorizer"
 	infra_authorizer "github.com/exanubes/appsync/internal/infrastructure/authorizer"
 	"github.com/exanubes/appsync/internal/infrastructure/clock"
 	"github.com/exanubes/appsync/internal/infrastructure/codec"
@@ -138,24 +139,32 @@ func (builder *builder) Connect(ctx context.Context) (*appsync_client, error) {
 	egress_queue := queue.NewEgressQueue(builder.backpressure.ConnectionOutbound)
 	pending_registry := pending.NewRegistry()
 	io_loops := io.New(ingress_queue, egress_queue, connection_output.Connection, msg_codec)
-	usecases := composition.NewUseCases(
+	usecases, services := composition.NewUseCases(
 		request_authorizer,
 		ingress_queue,
 		egress_queue,
 		pending_registry,
 		builder.backpressure.SubscriptionEvents,
 	)
+
 	msg_router := router.New(pending_registry, usecases.ReceiveData)
 	runtime := runtime.New(ingress_queue, msg_router, heartbeat)
 	session := engine.New(heartbeat, runtime, io_loops, builder.logger)
 	session.Start(ctx, engine.StartEngineInput{
 		Timeout: connection_output.Timeout,
 	})
 
+	shutdown_connection_usecase := shutdown.NewShutdownConnectionUsecase(
+		services.SubscriptionsRegistry,
+		services.RemoveSubscriptions,
+		session,
+		connection_output.Connection,
+	)
+
+	usecases.Shutdown = shutdown_connection_usecase
+
 	return &appsync_client{
-		transport: connection_output.Connection,
-		runtime:   session,
-		usecases:  usecases,
+		usecases: usecases,
 	}, nil
 }
 

internal/app/engine/engine.go

@@ -3,7 +3,6 @@ package engine
 import (
 	"context"
 	"errors"
-	"fmt"
 	"sync"
 
 	"github.com/exanubes/appsync/internal/app"
@@ -78,7 +77,7 @@ func (engine *Engine) Close(ctx context.Context) error {
 
 		case err := <-engine.err_channel:
 			if err != nil && !errors.Is(err, context.Canceled) {
-				error = fmt.Errorf("%w\n%w", error, err)
+				error = errors.Join(err)
 			}
 		default:
 		}

internal/app/services/connection/ports.go

@@ -50,9 +50,17 @@ type Writer interface {
 }
 
 type Closer interface {
-	Close() error
+	Close(context.Context) error
 }
 
 type Serializer interface {
 	Serialize(app.Signature) (string, error)
 }
+
+type Runtime interface {
+	Close(context.Context) error
+}
+
+type Transport interface {
+	Close(context.Context) error
+}

internal/app/services/subscription/create-subscription.go

@@ -5,11 +5,11 @@ import (
 )
 
 type CreateSubscriptionService struct {
-	registry     Registry
+	registry     SubscribeRegistry
 	backpressure uint
 }
 
-func NewCreateSubscriptionService(registry Registry, backpressure uint) *CreateSubscriptionService {
+func NewCreateSubscriptionService(registry SubscribeRegistry, backpressure uint) *CreateSubscriptionService {
 	return &CreateSubscriptionService{
 		registry:     registry,
 		backpressure: backpressure,

internal/app/services/subscription/ports.go

@@ -1,14 +1,35 @@
 package subscription
 
 import (
+	"context"
+
+	"github.com/exanubes/appsync/internal/app"
 	"github.com/exanubes/appsync/internal/app/subscription"
 )
 
-type Registry interface {
+type SubscribeRegistry interface {
 	Register(*subscription.Subscription)
 }
 
+type UnsubscribeRegistry interface {
+	Remove(id string)
+	Get(id string) *subscription.Subscription
+}
+
 type CreateSubscriptionInput struct {
 	ID      string
 	Channel string
 }
+
+type Subscriptions interface {
+	Active() []string
+	Get(string) *subscription.Subscription
+}
+
+type FrameFactory interface {
+	Unsubscribe() app.FrameBuilder
+}
+
+type Unsubscriber interface {
+	Unsubscribe(ctx context.Context, subscription_id string) error
+}

internal/app/services/subscription/remove-subscriptions.go

@@ -0,0 +1,47 @@
+package subscription
+
+import (
+	"context"
+	"errors"
+	"sync"
+)
+
+type RemoveSubscriptionsService struct {
+	subscription Unsubscriber
+}
+
+func NewRemoveSubscriptionsService(unsubscribe Unsubscriber) *RemoveSubscriptionsService {
+	return &RemoveSubscriptionsService{
+		subscription: unsubscribe,
+	}
+}
+
+func (service *RemoveSubscriptionsService) Remove(ctx context.Context, subscription_ids ...string) error {
+	var wg sync.WaitGroup
+	var err error
+	error_channel := make(chan error, len(subscription_ids))
+
+	for _, id := range subscription_ids {
+		wg.Go(func() {
+			select {
+			case error_channel <- service.subscription.Unsubscribe(ctx, id):
+			case <-ctx.Done():
+			default:
+			}
+		})
+	}
+	wg.Wait()
+	index := len(subscription_ids)
+	for index > 0 {
+		index -= 1
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case unsub_error := <-error_channel:
+			err = errors.Join(err, unsub_error)
+		default:
+		}
+	}
+
+	return err
+}

internal/app/services/subscription/unsubscribe.go

@@ -0,0 +1,61 @@
+package subscription
+
+import (
+	"context"
+	"github.com/exanubes/appsync/internal/app"
+	"github.com/exanubes/appsync/internal/app/protocol"
+)
+
+type UnsubscribeService struct {
+	subscriptions UnsubscribeRegistry
+	writer        app.SendMessageService
+	authorizer    app.RequestAuthorizer
+	frame         FrameFactory
+}
+
+func NewUnsubscribeService(
+	subscriptions UnsubscribeRegistry,
+	writer app.SendMessageService,
+	authorizer app.RequestAuthorizer,
+	frame FrameFactory,
+) *UnsubscribeService {
+	return &UnsubscribeService{
+		frame:         frame,
+		authorizer:    authorizer,
+		writer:        writer,
+		subscriptions: subscriptions,
+	}
+}
+
+func (service UnsubscribeService) Unsubscribe(ctx context.Context, subscription_id string) error {
+	frame := service.frame.Unsubscribe()
+	subscription := service.subscriptions.Get(subscription_id)
+
+	if subscription == nil {
+		return app.ErrSubscriptionClosed
+	}
+
+	if !subscription.Active() {
+		return app.ErrSubscriptionClosed
+	}
+
+	signature, err := service.authorizer.Authorize(ctx, app.AuthorizeCommandInput{})
+	if err != nil {
+		return err
+	}
+
+	frame.WithType(protocol.TypeUnsubscribe).
+		WithSignature(signature).
+		WithID(subscription_id)
+
+	err = service.writer.Send(ctx, frame.Build())
+
+	if err != nil {
+		return err
+	}
+
+	subscription.Close()
+	service.subscriptions.Remove(subscription_id)
+
+	return nil
+}

internal/app/subscription/registry.go

@@ -38,3 +38,18 @@ func (registry *Registry) Get(id string) *Subscription {
 
 	return nil
 }
+
+func (registry *Registry) Active() []string {
+	registry.mutex.RLock()
+	defer registry.mutex.RUnlock()
+
+	ids := []string{}
+
+	for id, sub := range registry.store {
+		if sub.Active() {
+			ids = append(ids, id)
+		}
+	}
+
+	return ids
+}

internal/app/usecases/shutdown/ports.go

@@ -0,0 +1,27 @@
+package shutdown
+
+import "context"
+
+type ShutdownConnection interface {
+	Execute(context.Context) error
+}
+
+type SubscriptionRegistry interface {
+	Active() []string
+}
+
+type Closer interface {
+	Close(context.Context) error
+}
+
+type Remover interface {
+	Remove(context.Context, ...string) error
+}
+
+type Runtime interface {
+	Close(context.Context) error
+}
+
+type Transport interface {
+	Close(context.Context) error
+}