consumer topic renamed to name

Author: hmoragrega

Makefile

@@ -26,9 +26,9 @@ test: up integration
 
 integration:
 	@mkdir -p coverage
-	@go test -race -v -tags=integration -coverpkg=./... -coverprofile=./coverage/base.cov ./...
-	@cd aws && go test -race -v -tags=integration -coverpkg=./... -coverprofile=../coverage/awsv2.cov ./...
-	@cd schedule/storage/postgres && go test -race -v -tags=integration -coverpkg=./... -coverprofile=../../../coverage/postgres.cov ./...
+	@go test -v -tags=integration -coverpkg=./... -coverprofile=./coverage/base.cov ./...
+	@cd aws && go test -v -tags=integration -coverpkg=./... -coverprofile=../coverage/awsv2.cov ./...
+	@cd schedule/storage/postgres && go test -v -tags=integration -coverpkg=./... -coverprofile=../../../coverage/postgres.cov ./...
 
 .PHONY: clean
 clean:

handler.go

@@ -24,7 +24,7 @@ func (f HandlerFunc) HandleMessage(ctx context.Context, message *Message) error
 }
 
 // Dispatcher is a message handler middleware that can be used to register
-// different handlers for the same topic, based on the message name.
+// different handlers for the same consumer, based on the message name.
 func Dispatcher(handlers map[string]Handler) HandlerFunc {
 	return func(ctx context.Context, message *Message) error {
 		h, ok := handlers[message.Name]

publisher.go

@@ -8,7 +8,7 @@ import (
 
 var (
 	// ErrResourceDoesNotExist indicates that the resource (topic, queue...) does not exist.
-	ErrResourceDoesNotExist = errors.New("topic does not exist")
+	ErrResourceDoesNotExist = errors.New("name does not exist")
 )
 
 // Publisher describes the top level method to publish messages.

publisher_recorder_test.go

@@ -9,7 +9,7 @@ func TestPublisherRecorder(t *testing.T) {
 	ctx := context.Background()
 	p := NewPublisherRecorder(NoOpPublisher())
 
-	err := p.Publish(ctx, "topic-1", &Message{
+	err := p.Publish(ctx, "name-1", &Message{
 		ID: "123",
 	}, &Message{
 		ID: "456",
@@ -18,7 +18,7 @@ func TestPublisherRecorder(t *testing.T) {
 		t.Fatal("unexpected error publishing", err)
 	}
 
-	err = p.Publish(ctx, "topic-2", &Message{
+	err = p.Publish(ctx, "name-2", &Message{
 		ID: "987",
 	}, &Message{
 		ID: "654",
@@ -35,27 +35,27 @@ func TestPublisherRecorder(t *testing.T) {
 		topic string
 		id    string
 	}{
-		{"topic-1", "123"},
-		{"topic-1", "456"},
-		{"topic-2", "987"},
-		{"topic-2", "654"},
+		{"name-1", "123"},
+		{"name-1", "456"},
+		{"name-2", "987"},
+		{"name-2", "654"},
 	} {
 		if got, want := p.Messages()[i].Topic, want.topic; got != want {
-			t.Fatalf("unexpected message topic at position %d; got %s, want %s", i, got, want)
+			t.Fatalf("unexpected message name at position %d; got %s, want %s", i, got, want)
 		}
 		if got, want := p.Messages()[i].Message.ID, want.id; got != want {
 			t.Fatalf("unexpected message id at position %d; got %s, want %s", i, got, want)
 		}
 	}
 
 	for i, want := range []string{"123", "456"} {
-		if got := p.MessagesMap()["topic-1"][i].ID; got != want {
+		if got := p.MessagesMap()["name-1"][i].ID; got != want {
 			t.Fatalf("unexpected message id at position %d; got %s, want %s", i, got, want)
 		}
 	}
 
 	for i, want := range []string{"987", "654"} {
-		if got := p.TopicMessages("topic-2")[i].ID; got != want {
+		if got := p.TopicMessages("name-2")[i].ID; got != want {
 			t.Fatalf("unexpected message id at position %d; got %s, want %s", i, got, want)
 		}
 	}

publisher_test.go

@@ -56,7 +56,7 @@ func TestPublisher_PublishHandlerSendMessages(t *testing.T) {
 		t.Fatalf("expected error; got %v, want %v", err, dummyErr)
 	}
 	if publishedTopic != topicToSend {
-		t.Fatalf("unpected pubslihing topic; got %v, want %v", publishedTopic, topicToSend)
+		t.Fatalf("unpected pubslihing name; got %v, want %v", publishedTopic, topicToSend)
 	}
 	if reflect.DeepEqual(messagesToSend, publishedMessages) {
 		t.Fatalf("expected error; got %v, want %v", err, dummyErr)
@@ -91,7 +91,7 @@ func TestPublisher_Wrapper(t *testing.T) {
 				t.Fatalf("message does not have the injected attribute; got %+v", e.Attributes)
 			}
 			if e.Name != topic {
-				t.Fatalf("topic should be used as event name; got %+v", e.Name)
+				t.Fatalf("name should be used as event name; got %+v", e.Name)
 			}
 		}
 		return nil
@@ -105,7 +105,7 @@ func TestPublisher_Wrapper(t *testing.T) {
 	)
 
 	ctx := context.WithValue(context.Background(), ctxKey, injectedValue)
-	err := p.Publish(ctx, "foo-topic", &pubsub.Message{})
+	err := p.Publish(ctx, "foo-name", &pubsub.Message{})
 	if err != nil {
 		t.Fatal("unexpected error publishing", err)
 	}

router.go

@@ -11,9 +11,9 @@ import (
 )
 
 var (
-	ErrTopicAlreadyRegistered = errors.New("topic already registered")
-	ErrRouterAlreadyRunning   = errors.New("router already running")
-	ErrRouterAlreadyStopped   = errors.New("router already stopped")
+	ErrConsumerAlreadyRegistered = errors.New("consumer already registered")
+	ErrRouterAlreadyRunning      = errors.New("router already running")
+	ErrRouterAlreadyStopped      = errors.New("router already stopped")
 )
 
 type status uint8
@@ -37,9 +37,9 @@ const (
 // Consumer consumes messages from a single subscription.
 type consumer struct {
 	subscriber Subscriber
-	handler    Handler
-	topic      string
-	next       <-chan Next
+	handler Handler
+	name    string
+	next    <-chan Next
 	backoff    BackoffStrategy
 	ackDecider AckDecider
 }
@@ -134,7 +134,7 @@ func WithBackoff(strategy BackoffStrategy) func(*consumer) {
 	}
 }
 
-func (r *Router) Register(topic string, subscriber Subscriber, handler Handler, opts ...ConsumerOption) error {
+func (r *Router) Register(name string, subscriber Subscriber, handler Handler, opts ...ConsumerOption) error {
 	r.mx.Lock()
 	defer r.mx.Unlock()
 
@@ -145,24 +145,24 @@ func (r *Router) Register(topic string, subscriber Subscriber, handler Handler,
 		return ErrRouterAlreadyStopped
 	}
 
-	_, found := r.consumers[topic]
+	_, found := r.consumers[name]
 	if found {
-		return fmt.Errorf("%w: %s", ErrTopicAlreadyRegistered, topic)
+		return fmt.Errorf("%w: %s", ErrConsumerAlreadyRegistered, name)
 	}
 
 	if r.consumers == nil {
 		r.consumers = make(map[string]*consumer)
 	}
 	c := &consumer{
-		topic:      topic,
+		name:       name,
 		subscriber: subscriber,
 		handler:    handler,
 	}
 	for _, opt := range opts {
 		opt(c)
 	}
 
-	r.consumers[topic] = c
+	r.consumers[name] = c
 
 	return nil
 }
@@ -211,7 +211,7 @@ func (r *Router) subscribe(consumers map[string]*consumer) ([]*consumer, error)
 		g.Go(func() error {
 			next, subscribeErr := c.subscriber.Subscribe()
 			if subscribeErr != nil {
-				subscribeErr = fmt.Errorf("subscribe to topic %s failed: %w", c.topic, subscribeErr)
+				subscribeErr = fmt.Errorf("subscribe to name %s failed: %w", c.name, subscribeErr)
 			}
 
 			if subscribeErr == nil {
@@ -249,7 +249,7 @@ func (r *Router) stop(consumers []*consumer) error {
 		g.Go(func() error {
 			stopErr := c.subscriber.Stop(ctx)
 			if stopErr != nil {
-				stopErr = fmt.Errorf("error stopping subscriber for topic %s: %w", c.topic, stopErr)
+				stopErr = fmt.Errorf("error stopping subscriber for consumer %s: %w", c.name, stopErr)
 			}
 			return stopErr
 		})
@@ -265,7 +265,7 @@ func (r *Router) run(ctx context.Context, consumers []*consumer) (err error) {
 		g.Go(func() error {
 			consumerErr := r.consume(ctx, c)
 			if consumerErr != nil {
-				consumerErr = fmt.Errorf("error consuming from topic %s: %w", c.topic, consumerErr)
+				consumerErr = fmt.Errorf("error consuming from consumer %s: %w", c.name, consumerErr)
 			}
 			return consumerErr
 		})
@@ -293,7 +293,7 @@ func (r *Router) consume(ctx context.Context, c *consumer) error {
 			continue
 		}
 
-		data, err := r.Unmarshaller.Unmarshal(c.topic, rmsg)
+		data, err := r.Unmarshaller.Unmarshal(c.name, rmsg)
 		if err := r.check(ctx, r.OnUnmarshal, c, rmsg, err); err != nil {
 			return err
 		}
@@ -325,7 +325,7 @@ func (r *Router) consume(ctx context.Context, c *consumer) error {
 
 func (r *Router) check(ctx context.Context, f Checkpoint, c *consumer, msg ReceivedMessage, err error) error {
 	if f != nil {
-		return f(ctx, c.topic, msg, err)
+		return f(ctx, c.name, msg, err)
 	}
 	return nil
 }
@@ -370,10 +370,10 @@ func (r *Router) messageContext(ctx context.Context, msg ReceivedMessage) contex
 
 func (r *Router) ack(ctx context.Context, c *consumer, msg ReceivedMessage, err error) Acknowledgement {
 	if c.ackDecider != nil {
-		return c.ackDecider(ctx, c.topic, msg, err)
+		return c.ackDecider(ctx, c.name, msg, err)
 	}
 
-	return r.AckDecider(ctx, c.topic, msg, err)
+	return r.AckDecider(ctx, c.name, msg, err)
 }
 
 func (r *Router) backoff(c *consumer, msg *Message) time.Duration {

router_test.go

@@ -31,17 +31,17 @@ func TestRouter_Run(t *testing.T) {
 		}
 	})
 
-	t.Run("cannot subscribe twice to the same topic", func(t *testing.T) {
+	t.Run("cannot subscribe twice using the same consumer name", func(t *testing.T) {
 		var router pubsub.Router
 
 		err := router.Register("same", &stubs.SubscriberStub{}, handlerDummy)
 		if err != nil {
-			t.Fatalf("unexpected error registering the topic: %v", err)
+			t.Fatalf("unexpected error registering the consumer: %v", err)
 		}
 
 		err = router.Register("same", &stubs.SubscriberStub{}, handlerDummy)
 		if err == nil {
-			t.Fatalf("unexpected success registering the topic again: %v", err)
+			t.Fatalf("unexpected success registering the consumer again: %v", err)
 		}
 	})
 
@@ -160,7 +160,7 @@ func TestRouter_Run(t *testing.T) {
 	})
 
 	t.Run("checkpoints are called", func(t *testing.T) {
-		subscriberTopic := "foo"
+		consumerName := "foo"
 
 		var checkpointsCalled int
 
@@ -171,31 +171,31 @@ func TestRouter_Run(t *testing.T) {
 			return nil
 		}
 
-		verifyCheckpoint := func(checkpoint string, topic string, message pubsub.ReceivedMessage, err error) error {
+		verifyCheckpoint := func(checkpoint string, name string, message pubsub.ReceivedMessage, err error) error {
 			checkpointsCalled++
 			if err != nil {
 				return err
 			}
 			if message.(*stubs.ReceivedMessageStub) != msg {
 				return fmt.Errorf("%s mesage is not the equal; got %+v", checkpoint, message)
 			}
-			if topic != subscriberTopic {
-				return fmt.Errorf("%s topic is not correct; got %+v, want %s", checkpoint, topic, subscriberTopic)
+			if name != consumerName {
+				return fmt.Errorf("%s consumer is not correct; got %+v, want %s", checkpoint, name, consumerName)
 			}
 			return nil
 		}
 		router := pubsub.Router{
-			OnReceive: func(_ context.Context, topic string, message pubsub.ReceivedMessage, err error) error {
-				return verifyCheckpoint("OnReceive", topic, message, err)
+			OnReceive: func(_ context.Context, name string, message pubsub.ReceivedMessage, err error) error {
+				return verifyCheckpoint("OnReceive", name, message, err)
 			},
-			OnUnmarshal: func(_ context.Context, topic string, message pubsub.ReceivedMessage, err error) error {
-				return verifyCheckpoint("OnUnmarshal", topic, message, err)
+			OnUnmarshal: func(_ context.Context, name string, message pubsub.ReceivedMessage, err error) error {
+				return verifyCheckpoint("OnUnmarshal", name, message, err)
 			},
-			OnHandler: func(_ context.Context, topic string, message pubsub.ReceivedMessage, err error) error {
-				return verifyCheckpoint("OnHandler", topic, message, err)
+			OnHandler: func(_ context.Context, name string, message pubsub.ReceivedMessage, err error) error {
+				return verifyCheckpoint("OnHandler", name, message, err)
 			},
-			OnAck: func(_ context.Context, topic string, message pubsub.ReceivedMessage, err error) error {
-				return verifyCheckpoint("OnAck", topic, message, err)
+			OnAck: func(_ context.Context, name string, message pubsub.ReceivedMessage, err error) error {
+				return verifyCheckpoint("OnAck", name, message, err)
 			},
 		}
 
@@ -211,7 +211,7 @@ func TestRouter_Run(t *testing.T) {
 			},
 		}
 
-		err := router.Register(subscriberTopic, s, handlerDummy)
+		err := router.Register(consumerName, s, handlerDummy)
 		if err != nil {
 			t.Fatal("cannot register handler", err)
 		}

scheduler_test.go

@@ -33,7 +33,7 @@ func TestPublisher(t *testing.T) {
 				return fmt.Errorf("due date is not in range: expected %v; got %v", expected, dueDate)
 			}
 			if gotTopic != topic {
-				return fmt.Errorf("unexpected topic: expected %v; got %v", topic, gotTopic)
+				return fmt.Errorf("unexpected name: expected %v; got %v", topic, gotTopic)
 			}
 			if len(envelopes) != 1 {
 				return fmt.Errorf("no messages sent")