test: unsubscribe

domain/entities/errors.go

@@ -7,4 +7,5 @@ var (
 	ErrTopicsNotFound        = errors.New("no topic was found")
 	ErrTopicNotFound         = errors.New("there isn't any subscriber listening to this topic")
 	ErrTopicAlreadyExists    = errors.New("this topic already exists")
+	ErrSubscriberNotFound    = errors.New("subscriber not found")
 )

domain/entities/topic.go

@@ -1,6 +1,7 @@
 package entities
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/matheusmosca/walrus/domain/vos"
@@ -31,7 +32,6 @@ func NewTopic(topicName vos.TopicName) (Topic, error) {
 func (t Topic) Activate() {
 	go t.listenForSubscriptions()
 	go t.listenForMessages()
-	go t.listenForKills()
 }
 
 func (t Topic) Dispatch(message vos.Message) error {
@@ -44,23 +44,37 @@ func (t Topic) Dispatch(message vos.Message) error {
 	return nil
 }
 
-func (t Topic) RemoveSubscriber(subscriberID vos.SubscriberID) {
-	t.killSubCh <- subscriberID
+func (t *Topic) RemoveSubscriber(subscriberID vos.SubscriberID) error {
+	if _, ok := t.subscribers.LoadAndDelete(subscriberID); !ok {
+		return ErrSubscriberNotFound
+	}
+
+	return nil
 }
 
 func (t Topic) addSubscriber(sub Subscriber) {
 	t.newSubCh <- sub
 }
 
-func (t *Topic) listenForSubscriptions() {
-	for newSubCh := range t.newSubCh {
-		t.subscribers.Store(newSubCh.GetID(), newSubCh)
+func (t Topic) GetSubscriber(subscriberID vos.SubscriberID) (*Subscriber, error) {
+	if value, ok := t.subscribers.Load(subscriberID); ok {
+		subsInterface := value.(Subscriber)
+
+		return &subsInterface, nil
 	}
+
+	return nil, ErrSubscriberNotFound
+}
+
+func (t Topic) UpdateTopic(topic Topic) (Topic, error) {
+	fmt.Println("implement me")
+
+	return Topic{}, nil
 }
 
-func (t *Topic) listenForKills() {
-	for subscriberID := range t.killSubCh {
-		t.subscribers.Delete(subscriberID)
+func (t *Topic) listenForSubscriptions() {
+	for newSubCh := range t.newSubCh {
+		t.subscribers.Store(newSubCh.GetID(), newSubCh)
 	}
 }
 

domain/usecases/unsubscribe.go

@@ -12,7 +12,12 @@ func (u useCase) Unsubscribe(ctx context.Context, subscriberID vos.SubscriberID,
 		return err
 	}
 
-	topic.RemoveSubscriber(subscriberID)
+	err = topic.RemoveSubscriber(subscriberID)
+	if err != nil {
+		return err
+	}
+
+	//u.storage.UpdateTopic(ctx, topic)
 
 	return nil
 }

domain/usecases/unsubscribe_test.go

@@ -0,0 +1,117 @@
+package usecases
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/matheusmosca/walrus/domain/entities"
+	"github.com/matheusmosca/walrus/domain/vos"
+)
+
+func TestUnsubscribe(t *testing.T) {
+	t.Parallel()
+
+	type args struct {
+		ctx     context.Context
+		message vos.Message
+	}
+
+	type fields struct {
+		storage   Repository
+		topicName vos.TopicName
+	}
+
+	tests := []struct {
+		name           string
+		args           args
+		fields         fields
+		beforeRun      func(topic entities.Topic) (chan vos.Message, vos.SubscriberID)
+		wantErr        error
+		wantSubscriber bool
+	}{
+		{
+			name: "unsubscribe should succeed",
+			args: args{
+				ctx: context.Background(),
+				message: vos.Message{
+					TopicName: "unsubscribing",
+				},
+			},
+			fields: fields{
+				storage:   &RepositoryMock{},
+				topicName: "unsubscribing",
+			},
+			beforeRun: func(topic entities.Topic) (chan vos.Message, vos.SubscriberID) {
+				subscriber := entities.NewSubscriber(topic)
+				ch, ID := subscriber.Subscribe()
+
+				// Assert that the subscriber has been created successfully
+				sub, err := topic.GetSubscriber(ID)
+				require.NoError(t, err)
+				assert.NotNil(t, sub)
+
+				return ch, ID
+			},
+			wantErr:        nil,
+			wantSubscriber: true,
+		},
+		{
+			name: "try to unsubscribe a subscriber that doesnt exist should fail",
+			args: args{
+				ctx: context.Background(),
+				message: vos.Message{
+					TopicName: "unsubscribing",
+				},
+			},
+			fields: fields{
+				storage:   &RepositoryMock{},
+				topicName: "unsubscribing",
+			},
+			beforeRun: func(topic entities.Topic) (chan vos.Message, vos.SubscriberID) {
+				msgChan := make(chan vos.Message)
+
+				return msgChan, vos.SubscriberID("")
+			},
+			wantErr:        entities.ErrSubscriberNotFound,
+			wantSubscriber: false,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			topic, err := entities.NewTopic(tt.fields.topicName)
+			require.NoError(t, err)
+
+			topic.Activate()
+
+			subsCh, subsID := tt.beforeRun(topic)
+			defer close(subsCh)
+
+			tt.fields.storage = &RepositoryMock{
+				GetTopicFunc: func(ctx context.Context, topicName vos.TopicName) (entities.Topic, error) {
+					if tt.fields.topicName != tt.args.message.TopicName {
+						return entities.Topic{}, entities.ErrTopicNotFound
+					}
+
+					return topic, nil
+				},
+			}
+
+			useCase := New(tt.fields.storage)
+			err = useCase.Unsubscribe(context.Background(), subsID, tt.fields.topicName)
+			if tt.wantErr != nil {
+				assert.ErrorIs(t, err, tt.wantErr)
+				return
+			}
+
+			// Assert the unsubscription
+			_, err = topic.GetSubscriber(subsID)
+			assert.ErrorIs(t, entities.ErrSubscriberNotFound, err)
+		})
+	}
+}

domain/usecases/usecase.go

@@ -18,6 +18,7 @@ type useCase struct {
 type Repository interface {
 	CreateTopic(ctx context.Context, name vos.TopicName, topic entities.Topic) error
 	GetTopic(ctx context.Context, topicName vos.TopicName) (entities.Topic, error)
+	UpdateTopic(ctx context.Context, topicName entities.Topic) (entities.Topic, error)
 	ListTopics(ctx context.Context) ([]entities.Topic, error)
 }
 

repositories/memory/topics/repository.go

@@ -17,6 +17,7 @@ type MemoryRepository interface {
 	GetTopic(ctx context.Context, topicName vos.TopicName) (entities.Topic, error)
 	CreateTopic(ctx context.Context, name vos.TopicName, topic entities.Topic) error
 	ListTopics(ctx context.Context) ([]entities.Topic, error)
+	UpdateTopic(ctx context.Context, topicName entities.Topic) (entities.Topic, error)
 }
 
 func NewMemoryRepository(storage map[vos.TopicName]entities.Topic) MemoryRepository {

repositories/memory/topics/update_topic.go

@@ -0,0 +1,11 @@
+package topics
+
+import (
+	"context"
+
+	"github.com/matheusmosca/walrus/domain/entities"
+)
+
+func (r *repository) UpdateTopic(ctx context.Context, topicName entities.Topic) (entities.Topic, error) {
+	return entities.Topic{}, nil
+}