willdot.net
An experimental pub/sub client and server project.
give each topic its own store
+29
-43
+1
-1
example/server/main.go
+1
-1
example/server/main.go
+1
-12
pubsub/subscriber_test.go
+1
-12
pubsub/subscriber_test.go
···
18
topicB = "topic b"
19
)
20
21
-
type fakeStore struct {
22
-
}
23
-
24
-
func (f *fakeStore) Write(msg server.MessageToSend) error {
25
-
return nil
26
-
}
27
-
func (f *fakeStore) ReadFrom(offset int, handleFunc func(msgs []server.MessageToSend)) error {
28
-
return nil
29
-
}
30
-
31
func createServer(t *testing.T) {
32
-
fs := &fakeStore{}
33
-
server, err := server.New(serverAddr, time.Millisecond*100, time.Millisecond*100, fs)
34
require.NoError(t, err)
35
36
t.Cleanup(func() {
+2
-5
server/message_store.go
+2
-5
server/message_store.go
···
28
return nil
29
}
30
31
-
func (m *MemoryStore) ReadFrom(offset int, handleFunc func(msgs []MessageToSend)) error {
32
if offset < 0 || offset > m.offset {
33
return fmt.Errorf("invalid offset provided")
34
}
···
36
m.mu.Lock()
37
defer m.mu.Unlock()
38
39
-
msgs := make([]MessageToSend, 0, len(m.msgs))
40
for i := offset; i < len(m.msgs); i++ {
41
-
msgs = append(msgs, m.msgs[i])
42
}
43
-
44
-
handleFunc(msgs)
45
46
return nil
47
}
···
28
return nil
29
}
30
31
+
func (m *MemoryStore) ReadFrom(offset int, handleFunc func(msg MessageToSend)) error {
32
if offset < 0 || offset > m.offset {
33
return fmt.Errorf("invalid offset provided")
34
}
···
36
m.mu.Lock()
37
defer m.mu.Unlock()
38
39
for i := offset; i < len(m.msgs); i++ {
40
+
handleFunc(m.msgs[i])
41
}
42
43
return nil
44
}
+9
-12
server/server.go
+9
-12
server/server.go
···
77
78
type Store interface {
79
Write(msg MessageToSend) error
80
-
ReadFrom(offset int, handleFunc func(msgs []MessageToSend)) error
81
}
82
83
// Server accepts subscribe and publish connections and passes messages around
···
90
91
ackDelay time.Duration
92
ackTimeout time.Duration
93
-
94
-
messageStore Store
95
}
96
97
// New creates and starts a new server
98
-
func New(Addr string, ackDelay, ackTimeout time.Duration, messageStore Store) (*Server, error) {
99
lis, err := net.Listen("tcp", Addr)
100
if err != nil {
101
return nil, fmt.Errorf("failed to listen: %w", err)
102
}
103
104
srv := &Server{
105
-
lis: lis,
106
-
topics: map[string]*topic{},
107
-
ackDelay: ackDelay,
108
-
ackTimeout: ackTimeout,
109
-
messageStore: messageStore,
110
}
111
112
go srv.start()
···
375
376
topic := s.getTopic(message.topic)
377
if topic == nil {
378
-
topic = newTopic(message.topic, s.messageStore)
379
s.topics[message.topic] = topic
380
}
381
···
406
407
t, ok := s.topics[topicName]
408
if !ok {
409
-
t = newTopic(topicName, s.messageStore)
410
}
411
412
-
t.subscriptions[peer.Addr()] = newSubscriber(peer, topicName, s.ackDelay, s.ackTimeout, s.messageStore, startAt)
413
414
s.topics[topicName] = t
415
}
···
77
78
type Store interface {
79
Write(msg MessageToSend) error
80
+
ReadFrom(offset int, handleFunc func(msg MessageToSend)) error
81
}
82
83
// Server accepts subscribe and publish connections and passes messages around
···
90
91
ackDelay time.Duration
92
ackTimeout time.Duration
93
}
94
95
// New creates and starts a new server
96
+
func New(Addr string, ackDelay, ackTimeout time.Duration) (*Server, error) {
97
lis, err := net.Listen("tcp", Addr)
98
if err != nil {
99
return nil, fmt.Errorf("failed to listen: %w", err)
100
}
101
102
srv := &Server{
103
+
lis: lis,
104
+
topics: map[string]*topic{},
105
+
ackDelay: ackDelay,
106
+
ackTimeout: ackTimeout,
107
}
108
109
go srv.start()
···
372
373
topic := s.getTopic(message.topic)
374
if topic == nil {
375
+
topic = newTopic(message.topic)
376
s.topics[message.topic] = topic
377
}
378
···
403
404
t, ok := s.topics[topicName]
405
if !ok {
406
+
t = newTopic(topicName)
407
}
408
409
+
t.subscriptions[peer.Addr()] = newSubscriber(peer, topicName, s.ackDelay, s.ackTimeout, t.messageStore, startAt)
410
411
s.topics[topicName] = t
412
}
+12
-5
server/server_test.go
+12
-5
server/server_test.go
···
24
)
25
26
func createServer(t *testing.T) *Server {
27
-
store := NewMemoryStore()
28
-
srv, err := New(serverAddr, ackDelay, ackTimeout, store)
29
require.NoError(t, err)
30
31
t.Cleanup(func() {
···
40
srv.topics[topicName] = &topic{
41
name: topicName,
42
subscriptions: make(map[net.Addr]*subscriber),
43
}
44
45
return srv
···
516
messages = append(messages, fmt.Sprintf("message %d", i))
517
}
518
519
-
// send messages first
520
topic := fmt.Sprintf("topic:%s", topicA)
521
522
-
// send multiple messages
523
for _, msg := range messages {
524
sendMessage(t, publisherConn, topic, []byte(msg))
525
}
526
527
subscriberConn := createConnectionAndSubscribe(t, []string{topicA}, From, 0)
528
results := make([]string, 0, len(messages))
···
547
messages = append(messages, fmt.Sprintf("message %d", i))
548
}
549
550
-
// send messages first
551
topic := fmt.Sprintf("topic:%s", topicA)
552
553
// send multiple messages
554
for _, msg := range messages {
555
sendMessage(t, publisherConn, topic, []byte(msg))
556
}
557
558
subscriberConn := createConnectionAndSubscribe(t, []string{topicA}, From, 3)
559
···
24
)
25
26
func createServer(t *testing.T) *Server {
27
+
srv, err := New(serverAddr, ackDelay, ackTimeout)
28
require.NoError(t, err)
29
30
t.Cleanup(func() {
···
39
srv.topics[topicName] = &topic{
40
name: topicName,
41
subscriptions: make(map[net.Addr]*subscriber),
42
+
messageStore: NewMemoryStore(),
43
}
44
45
return srv
···
516
messages = append(messages, fmt.Sprintf("message %d", i))
517
}
518
519
topic := fmt.Sprintf("topic:%s", topicA)
520
521
for _, msg := range messages {
522
sendMessage(t, publisherConn, topic, []byte(msg))
523
}
524
+
525
+
// send some messages for topic B as well
526
+
sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 1"))
527
+
sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 2"))
528
+
sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 3"))
529
530
subscriberConn := createConnectionAndSubscribe(t, []string{topicA}, From, 0)
531
results := make([]string, 0, len(messages))
···
550
messages = append(messages, fmt.Sprintf("message %d", i))
551
}
552
553
topic := fmt.Sprintf("topic:%s", topicA)
554
555
// send multiple messages
556
for _, msg := range messages {
557
sendMessage(t, publisherConn, topic, []byte(msg))
558
}
559
+
560
+
// send some messages for topic B as well
561
+
sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 1"))
562
+
sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 2"))
563
+
sendMessage(t, publisherConn, fmt.Sprintf("topic:%s", topicB), []byte("topic b message 3"))
564
565
subscriberConn := createConnectionAndSubscribe(t, []string{topicA}, From, 3)
566
+2
-7
server/subscriber.go
+2
-7
server/subscriber.go
···
44
offset := startAt
45
46
go func() {
47
-
// here we need to replay all messages from the store for the topic.
48
-
err := messageStore.ReadFrom(offset, func(msgs []MessageToSend) {
49
-
// go func() {
50
-
for _, msg := range msgs {
51
-
s.messages <- newMessage(msg.data)
52
-
}
53
-
// }()
54
})
55
if err != nil {
56
slog.Error("failed to replay messages from offset", "error", err, "offset", offset)
+2
-1
server/topic.go
+2
-1
server/topic.go