From 131bbe2c7605360b6062879df7da983d87b9ca8f Mon Sep 17 00:00:00 2001 From: Will Date: Wed, 7 Feb 2024 20:24:48 +0000 Subject: [PATCH] give each topic its own store --- example/server/main.go | 2 +- pubsub/subscriber_test.go | 13 +------------ server/message_store.go | 7 ++----- server/server.go | 21 +++++++++------------ server/server_test.go | 17 ++++++++++++----- server/subscriber.go | 9 ++------- server/topic.go | 3 ++- 7 files changed, 29 insertions(+), 43 deletions(-) diff --git a/example/server/main.go b/example/server/main.go index 4dece49..3841b19 100644 --- a/example/server/main.go +++ b/example/server/main.go @@ -11,7 +11,7 @@ import ( ) func main() { - srv, err := server.New(":3000", time.Second, time.Second*2, server.NewMemoryStore()) + srv, err := server.New(":3000", time.Second, time.Second*2) if err != nil { log.Fatal(err) } diff --git a/pubsub/subscriber_test.go b/pubsub/subscriber_test.go index a1c0b84..eff376e 100644 --- a/pubsub/subscriber_test.go +++ b/pubsub/subscriber_test.go @@ -18,19 +18,8 @@ const ( topicB = "topic b" ) -type fakeStore struct { -} - -func (f *fakeStore) Write(msg server.MessageToSend) error { - return nil -} -func (f *fakeStore) ReadFrom(offset int, handleFunc func(msgs []server.MessageToSend)) error { - return nil -} - func createServer(t *testing.T) { - fs := &fakeStore{} - server, err := server.New(serverAddr, time.Millisecond*100, time.Millisecond*100, fs) + server, err := server.New(serverAddr, time.Millisecond*100, time.Millisecond*100) require.NoError(t, err) t.Cleanup(func() { diff --git a/server/message_store.go b/server/message_store.go index d936e1e..8ebf768 100644 --- a/server/message_store.go +++ b/server/message_store.go @@ -28,7 +28,7 @@ func (m *MemoryStore) Write(msg MessageToSend) error { return nil } -func (m *MemoryStore) ReadFrom(offset int, handleFunc func(msgs []MessageToSend)) error { +func (m *MemoryStore) ReadFrom(offset int, handleFunc func(msg MessageToSend)) error { if offset < 0 || offset > m.offset { return fmt.Errorf("invalid offset provided") } @@ -36,12 +36,9 @@ func (m *MemoryStore) ReadFrom(offset int, handleFunc func(msgs []MessageToSend) m.mu.Lock() defer m.mu.Unlock() - msgs := make([]MessageToSend, 0, len(m.msgs)) for i := offset; i < len(m.msgs); i++ { - msgs = append(msgs, m.msgs[i]) + handleFunc(m.msgs[i]) } - handleFunc(msgs) - return nil } diff --git a/server/server.go b/server/server.go index d9b4ed3..5600f38 100644 --- a/server/server.go +++ b/server/server.go @@ -77,7 +77,7 @@ const ( type Store interface { Write(msg MessageToSend) error - ReadFrom(offset int, handleFunc func(msgs []MessageToSend)) error + ReadFrom(offset int, handleFunc func(msg MessageToSend)) error } // Server accepts subscribe and publish connections and passes messages around @@ -90,23 +90,20 @@ type Server struct { ackDelay time.Duration ackTimeout time.Duration - - messageStore Store } // New creates and starts a new server -func New(Addr string, ackDelay, ackTimeout time.Duration, messageStore Store) (*Server, error) { +func New(Addr string, ackDelay, ackTimeout time.Duration) (*Server, error) { lis, err := net.Listen("tcp", Addr) if err != nil { return nil, fmt.Errorf("failed to listen: %w", err) } srv := &Server{ - lis: lis, - topics: map[string]*topic{}, - ackDelay: ackDelay, - ackTimeout: ackTimeout, - messageStore: messageStore, + lis: lis, + topics: map[string]*topic{}, + ackDelay: ackDelay, + ackTimeout: ackTimeout, } go srv.start() @@ -375,7 +372,7 @@ func (s *Server) handlePublish(peer *peer.Peer) { topic := s.getTopic(message.topic) if topic == nil { - topic = newTopic(message.topic, s.messageStore) + topic = newTopic(message.topic) s.topics[message.topic] = topic } @@ -406,10 +403,10 @@ func (s *Server) addSubsciberToTopic(topicName string, peer *peer.Peer, startAt t, ok := s.topics[topicName] if !ok { - t = newTopic(topicName, s.messageStore) + t = newTopic(topicName) } - t.subscriptions[peer.Addr()] = newSubscriber(peer, topicName, s.ackDelay, s.ackTimeout, s.messageStore, startAt) + t.subscriptions[peer.Addr()] = newSubscriber(peer, topicName, s.ackDelay, s.ackTimeout, t.messageStore, startAt) s.topics[topicName] = t } diff --git a/server/server_test.go b/server/server_test.go index 3a836c2..e9deaf1 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -24,8 +24,7 @@ const ( ) func createServer(t *testing.T) *Server { - store := NewMemoryStore() - srv, err := New(serverAddr, ackDelay, ackTimeout, store) + srv, err := New(serverAddr, ackDelay, ackTimeout) require.NoError(t, err) t.Cleanup(func() { @@ -40,6 +39,7 @@ func createServerWithExistingTopic(t *testing.T, topicName string) *Server { srv.topics[topicName] = &topic{ name: topicName, subscriptions: make(map[net.Addr]*subscriber), + messageStore: NewMemoryStore(), } return srv @@ -516,14 +516,17 @@ func TestSubscribeAndReplaysFromStart(t *testing.T) { messages = append(messages, fmt.Sprintf("message %d", i)) } - // send messages first topic := fmt.Sprintf("topic:%s", topicA) - // send multiple messages for _, msg := range messages { sendMessage(t, publisherConn, topic, []byte(msg)) } + // send some messages for topic B as well + sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 1")) + sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 2")) + sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 3")) + subscriberConn := createConnectionAndSubscribe(t, []string{topicA}, From, 0) results := make([]string, 0, len(messages)) for i := 0; i < len(messages); i++ { @@ -547,7 +550,6 @@ func TestSubscribeAndReplaysFromIndex(t *testing.T) { messages = append(messages, fmt.Sprintf("message %d", i)) } - // send messages first topic := fmt.Sprintf("topic:%s", topicA) // send multiple messages @@ -555,6 +557,11 @@ func TestSubscribeAndReplaysFromIndex(t *testing.T) { sendMessage(t, publisherConn, topic, []byte(msg)) } + // send some messages for topic B as well + sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 1")) + sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 2")) + sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 3")) + subscriberConn := createConnectionAndSubscribe(t, []string{topicA}, From, 3) // now that the subscriber has subecribed send another message that should arrive after all the other messages were consumed diff --git a/server/subscriber.go b/server/subscriber.go index 56360d1..318994f 100644 --- a/server/subscriber.go +++ b/server/subscriber.go @@ -44,13 +44,8 @@ func newSubscriber(peer *peer.Peer, topic string, ackDelay, ackTimeout time.Dura offset := startAt go func() { - // here we need to replay all messages from the store for the topic. - err := messageStore.ReadFrom(offset, func(msgs []MessageToSend) { - // go func() { - for _, msg := range msgs { - s.messages <- newMessage(msg.data) - } - // }() + err := messageStore.ReadFrom(offset, func(msg MessageToSend) { + s.messages <- newMessage(msg.data) }) if err != nil { slog.Error("failed to replay messages from offset", "error", err, "offset", offset) diff --git a/server/topic.go b/server/topic.go index c5d255a..f9a9236 100644 --- a/server/topic.go +++ b/server/topic.go @@ -13,7 +13,8 @@ type topic struct { messageStore Store } -func newTopic(name string, messageStore Store) *topic { +func newTopic(name string) *topic { + messageStore := NewMemoryStore() return &topic{ name: name, subscriptions: make(map[net.Addr]*subscriber), -- 2.51.2