adjust

Author: anbraten

services/pubsub/memory.go

@@ -21,10 +21,10 @@ func NewMemory() Broker {
 	}
 }
 
-func (p *Memory) Publish(_ context.Context, message Message) {
+func (p *Memory) Publish(_ context.Context, _topic string, message Message) {
 	p.Lock()
 
-	topic, ok := p.topics[message.Topic]
+	topic, ok := p.topics[_topic]
 	if !ok {
 		p.Unlock()
 		return

services/pubsub/memory_test.go

@@ -17,8 +17,7 @@ func TestPubsub(t *testing.T) {
 		wg sync.WaitGroup
 
 		testMessage = Message{
-			Data:  []byte("test"),
-			Topic: "hello-world",
+			Data: []byte("test"),
 		}
 	)
 
@@ -39,7 +38,7 @@ func TestPubsub(t *testing.T) {
 
 	wg.Add(2)
 	go func() {
-		broker.Publish(ctx, testMessage)
+		broker.Publish(ctx, "hello-world", testMessage)
 	}()
 
 	wg.Wait()

services/pubsub/types.go

@@ -9,15 +9,12 @@ import "context"
 type Message struct {
 	// Data is the actual data in the entry.
 	Data []byte `json:"data"`
-
-	// Topic is the topic of the message.
-	Topic string `json:"topic"`
 }
 
 // Subscriber receives published messages.
 type Subscriber func(Message)
 
 type Broker interface {
-	Publish(c context.Context, message Message)
+	Publish(c context.Context, topic string, message Message)
 	Subscribe(c context.Context, topic string, subscriber Subscriber)
 }

services/websocket/issue_comment_notifier.go

@@ -19,8 +19,8 @@ func (n *websocketNotifier) DeleteComment(ctx context.Context, doer *user_model.
 		return
 	}
 
-	n.pubsub.Publish(ctx, pubsub.Message{
-		Data:  d,
-		Topic: fmt.Sprintf("repo:%s/%s", c.RefRepo.OwnerName, c.RefRepo.Name),
+	topic := fmt.Sprintf("repo:%s/%s", c.RefRepo.OwnerName, c.RefRepo.Name)
+	n.pubsub.Publish(ctx, topic, pubsub.Message{
+		Data: d,
 	})
 }