An experimental pub/sub client and server project.
Merge pull request #4 from willdot/ci
Add Ci and fix all the lint
authored by
GitHub
f5528bdb
71cb555e
+88
-36
+26
.github/workflows/workflow.yaml
+26
.github/workflows/workflow.yaml
···
1
+
name: Go package
2
+
3
+
on: [push]
4
+
5
+
jobs:
6
+
build:
7
+
runs-on: ubuntu-latest
8
+
steps:
9
+
- uses: actions/checkout@v3
10
+
11
+
- name: Set up Go
12
+
uses: actions/setup-go@v4
13
+
with:
14
+
go-version: '1.21'
15
+
16
+
- name: golangci-lint
17
+
run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin
18
+
19
+
- name: Build
20
+
run: go build -v ./...
21
+
22
+
- name: lint
23
+
run: golangci-lint run
24
+
25
+
- name: Test
26
+
run: go test ./... -p 1 -count=1 -v
+13
-4
example/main.go
+13
-4
example/main.go
···
16
16
consumeOnly = flag.Bool("consume-only", false, "just consumes (doesn't start server and doesn't publish)")
17
17
flag.Parse()
18
18
19
-
if *consumeOnly == false {
19
+
if !*consumeOnly {
20
20
go sendMessages()
21
21
}
22
22
···
24
24
if err != nil {
25
25
panic(err)
26
26
}
27
-
defer sub.Close()
27
+
28
+
defer func() {
29
+
_ = sub.Close()
30
+
}()
28
31
29
-
sub.SubscribeToTopics([]string{"topic a"})
32
+
err = sub.SubscribeToTopics([]string{"topic a"})
33
+
if err != nil {
34
+
panic(err)
35
+
}
30
36
31
37
consumer := sub.Consume(context.Background())
32
38
if consumer.Err != nil {
···
44
50
if err != nil {
45
51
panic(err)
46
52
}
47
-
defer publisher.Close()
53
+
54
+
defer func() {
55
+
_ = publisher.Close()
56
+
}()
48
57
49
58
// send some messages
50
59
i := 0
+4
-1
example/server/main.go
+4
-1
example/server/main.go
+1
-1
pubsub/subscriber_test.go
+1
-1
pubsub/subscriber_test.go
+4
-4
server/peer/peer.go
+4
-4
server/peer/peer.go
···
5
5
"sync"
6
6
)
7
7
8
-
// Peer represents a remote connection to the server such as a publisher or subscriber
8
+
// Peer represents a remote connection to the server such as a publisher or subscriber.
9
9
type Peer struct {
10
10
conn net.Conn
11
11
connMu sync.Mutex
12
12
}
13
13
14
-
// New returns a new peer
14
+
// New returns a new peer.
15
15
func New(conn net.Conn) *Peer {
16
16
return &Peer{
17
17
conn: conn,
18
18
}
19
19
}
20
20
21
-
// Addr returns the peers connections address
21
+
// Addr returns the peers connections address.
22
22
func (p *Peer) Addr() net.Addr {
23
23
return p.conn.RemoteAddr()
24
24
}
25
25
26
-
// ConnOpp represents a set of actions on a connection that can be used synchrnously
26
+
// ConnOpp represents a set of actions on a connection that can be used synchrnously.
27
27
type ConnOpp func(conn net.Conn) error
28
28
29
29
// RunConnOperation will run the provided operation. It ensures that it is the only operation that is being
+30
-14
server/server.go
+30
-14
server/server.go
···
5
5
"encoding/json"
6
6
"errors"
7
7
"fmt"
8
+
"io"
8
9
"log/slog"
9
10
"net"
10
11
"strings"
11
12
"sync"
13
+
"syscall"
12
14
"time"
13
15
14
16
"github.com/willdot/messagebroker/server/peer"
···
51
53
lis net.Listener
52
54
53
55
mu sync.Mutex
54
-
topics map[string]topic
56
+
topics map[string]*topic
55
57
}
56
58
57
59
// New creates and starts a new server
···
63
65
64
66
srv := &Server{
65
67
lis: lis,
66
-
topics: map[string]topic{},
68
+
topics: map[string]*topic{},
67
69
}
68
70
69
71
go srv.start()
···
97
99
98
100
action, err := readAction(peer, 0)
99
101
if err != nil {
100
-
slog.Error("failed to read action from peer", "error", err, "peer", peer.Addr())
102
+
if !errors.Is(err, io.EOF) {
103
+
slog.Error("failed to read action from peer", "error", err, "peer", peer.Addr())
104
+
}
101
105
return
102
106
}
103
107
···
123
127
for {
124
128
action, err := readAction(peer, time.Millisecond*100)
125
129
if err != nil {
130
+
// if the error is a timeout, it means the peer hasn't sent an action indicating it wishes to do something so sleep
131
+
// for a little bit to allow for other actions to happen on the connection
126
132
var neterr net.Error
127
133
if errors.As(err, &neterr) && neterr.Timeout() {
128
-
time.Sleep(time.Second)
134
+
time.Sleep(time.Millisecond * 500)
129
135
continue
130
136
}
131
-
// TODO: see if there's a way to check if the peers connection has been ended etc
132
-
slog.Error("failed to read action from subscriber", "error", err, "peer", peer.Addr())
137
+
138
+
if !errors.Is(err, io.EOF) {
139
+
slog.Error("failed to read action from subscriber", "error", err, "peer", peer.Addr())
140
+
}
133
141
134
-
s.unsubscribePeerFromAllTopics(*peer)
142
+
s.unsubscribePeerFromAllTopics(peer)
135
143
136
144
return
137
145
}
···
218
226
return nil
219
227
}
220
228
221
-
s.unsubscribeToTopics(*peer, topics)
229
+
s.unsubscribeToTopics(peer, topics)
222
230
writeStatus(Unsubscribed, "", conn)
223
231
224
232
return nil
···
239
247
op := func(conn net.Conn) error {
240
248
dataLen, err := dataLength(conn)
241
249
if err != nil {
250
+
if errors.Is(err, io.EOF) {
251
+
return nil
252
+
}
242
253
slog.Error("failed to read data length", "error", err, "peer", peer.Addr())
243
254
writeStatus(Error, "invalid data length of data provided", conn)
244
255
return nil
···
325
336
s.topics[topicName] = t
326
337
}
327
338
328
-
func (s *Server) unsubscribeToTopics(peer peer.Peer, topics []string) {
339
+
func (s *Server) unsubscribeToTopics(peer *peer.Peer, topics []string) {
329
340
for _, topic := range topics {
330
341
s.removeSubsciberFromTopic(topic, peer)
331
342
}
332
343
}
333
344
334
-
func (s *Server) removeSubsciberFromTopic(topicName string, peer peer.Peer) {
345
+
func (s *Server) removeSubsciberFromTopic(topicName string, peer *peer.Peer) {
335
346
s.mu.Lock()
336
347
defer s.mu.Unlock()
337
348
···
343
354
delete(t.subscriptions, peer.Addr())
344
355
}
345
356
346
-
func (s *Server) unsubscribePeerFromAllTopics(peer peer.Peer) {
357
+
func (s *Server) unsubscribePeerFromAllTopics(peer *peer.Peer) {
347
358
s.mu.Lock()
348
359
defer s.mu.Unlock()
349
360
···
357
368
defer s.mu.Unlock()
358
369
359
370
if topic, ok := s.topics[topicName]; ok {
360
-
return &topic
371
+
return topic
361
372
}
362
373
363
374
return nil
···
367
378
var action Action
368
379
op := func(conn net.Conn) error {
369
380
if timeout > 0 {
370
-
conn.SetReadDeadline(time.Now().Add(timeout))
381
+
err := conn.SetReadDeadline(time.Now().Add(timeout))
382
+
if err != nil {
383
+
slog.Error("failed to set connection read deadline", "error", err, "peer", peer.Addr())
384
+
}
371
385
}
372
386
373
387
err := binary.Read(conn, binary.BigEndian, &action)
···
406
420
func writeStatus(status Status, message string, conn net.Conn) {
407
421
err := binary.Write(conn, binary.BigEndian, status)
408
422
if err != nil {
409
-
slog.Error("failed to write status to peers connection", "error", err, "peer", conn.RemoteAddr())
423
+
if !errors.Is(err, syscall.EPIPE) {
424
+
slog.Error("failed to write status to peers connection", "error", err, "peer", conn.RemoteAddr())
425
+
}
410
426
return
411
427
}
412
428
+8
-2
server/server_test.go
+8
-2
server/server_test.go
···
25
25
require.NoError(t, err)
26
26
27
27
t.Cleanup(func() {
28
-
srv.Shutdown()
28
+
_ = srv.Shutdown()
29
29
})
30
30
31
31
return srv
···
33
33
34
34
func createServerWithExistingTopic(t *testing.T, topicName string) *Server {
35
35
srv := createServer(t)
36
-
srv.topics[topicName] = topic{
36
+
srv.topics[topicName] = &topic{
37
37
name: topicName,
38
38
subscriptions: make(map[net.Addr]subscriber),
39
39
}
···
61
61
62
62
var resp Status
63
63
err = binary.Read(conn, binary.BigEndian, &resp)
64
+
require.NoError(t, err)
64
65
65
66
assert.Equal(t, expectedRes, int(resp))
66
67
···
106
107
107
108
var resp Status
108
109
err = binary.Read(conn, binary.BigEndian, &resp)
110
+
require.NoError(t, err)
109
111
110
112
assert.Equal(t, expectedRes, int(resp))
111
113
···
160
162
161
163
var resp Status
162
164
err = binary.Read(conn, binary.BigEndian, &resp)
165
+
require.NoError(t, err)
163
166
164
167
assert.Equal(t, expectedRes, int(resp))
165
168
···
167
170
168
171
var dataLen uint32
169
172
err = binary.Read(conn, binary.BigEndian, &dataLen)
173
+
require.NoError(t, err)
170
174
assert.Equal(t, len(expectedMessage), int(dataLen))
171
175
172
176
buf := make([]byte, dataLen)
···
196
200
197
201
var resp Status
198
202
err = binary.Read(publisherConn, binary.BigEndian, &resp)
203
+
require.NoError(t, err)
199
204
200
205
assert.Equal(t, expectedRes, int(resp))
201
206
···
203
208
204
209
var dataLen uint32
205
210
err = binary.Read(publisherConn, binary.BigEndian, &dataLen)
211
+
require.NoError(t, err)
206
212
assert.Equal(t, len(expectedMessage), int(dataLen))
207
213
208
214
buf := make([]byte, dataLen)
+2
-10
server/topic.go
+2
-10
server/topic.go
···
21
21
currentOffset int
22
22
}
23
23
24
-
func newTopic(name string) topic {
25
-
return topic{
24
+
func newTopic(name string) *topic {
25
+
return &topic{
26
26
name: name,
27
27
subscriptions: make(map[net.Addr]subscriber),
28
28
}
29
-
}
30
-
31
-
func (t *topic) removeSubscriber(addr net.Addr) {
32
-
t.mu.Lock()
33
-
defer t.mu.Unlock()
34
-
35
-
slog.Info("removing subscriber", "peer", addr)
36
-
delete(t.subscriptions, addr)
37
29
}
38
30
39
31
func (t *topic) sendMessageToSubscribers(msgData []byte) {