alash3al/go-pubsub

An in-process simple publish/subscribe library for golang

Go Pub/Sub

An in-process simple publish/subscribe library for Golang .

Why

I wrote this package so I can use it in my projects as a drop-in-replacement for redis pub/sub, actually this package was written for wsify v2

Contribution

I'm always open for any opinion

License

I'm using MIT license.

Known Users

• Wsify
• RedixDB

go-pubsub's Issues

Be great to have an example.go for Newbies

Hey just found your package and just trying to work it out. Be great for a simple example.go file that would be a bit help in understanding how to use this package quickly. :)

Memory Leak

There's a memory leak.

func (b *Broker) Attach ...
b.subscribers[s.id] = s

s is never removed from the b.subscribers, and the gc will not release the object.
BTW. Obviously b.subscribers is superfluous since it is not referenced anywhere.

regards Kay

missing unlock?

hello - should you unlock before the continue, here:

go-pubsub/broker.go

Line 62 in 6961c9e

continue

??

cheers

Empty Topics

Currently when a topic becomes empty (all subscribers are unsubscribed), the topic is not removed from the topics map in the Broker struct.

Similarly, when a subscriber is detached from the broker it is:

1. Marked as destroyed and channel is closed
2. Unsubscribes from all topics, which goes back to my first case (empty topics aren't removed)

Why aren't empty topics being removed from the Broker?

Race condition

Ran into a race condition.
Seems to happen here: https://github.com/alash3al/go-pubsub/blob/master/broker.go#L88
Reference: https://tip.golang.org/doc/go1.8#mapiter

Is there a valid reason the Broadcast function got no mutex/lock?

fatal error: concurrent map iteration and map write
goroutine 127519 [running]:
runtime.throw(0xa481a5, 0x26)
/usr/lib/golang/src/runtime/panic.go:617 +0x72 fp=0xc000cc7838 sp=0xc000cc7808 pc=0x42f2b2
runtime.mapiternext(0xc000cc7920)
/usr/lib/golang/src/runtime/map.go:860 +0x597 fp=0xc000cc78c0 sp=0xc000cc7838 pc=0x410967
github.com/alash3al/go-pubsub.(*Broker).Broadcast(0xc000529d00, 0x9571e0, 0xc000c440b0, 0xc000c440c0, 0x1, 0x1)
/home/xxx/go/pkg/mod/github.com/alash3al/[email protected]/broker.go:88 +0x17d fp=0xc000cc7990 sp=0xc000>
ppp/cws/pkg/queue.PublishToWebsocketClose(...)
/home/xxx/yyy/zzz/go/src/ppp/cws/pkg/queue/pubsub,.go:38
ppp/cws/pkg/websocket.DockerConnection.func1(0xc000651080, 0xc0006da360, 0x24)
/home/xxx/yyy/zzz/go/src/ppp/cws/pkg/websocket/docker.go:52 +0x111 fp=0xc000cc79e8 sp=0xc000cc>
ppp/cws/pkg/websocket.DockerConnection(0xb45200, 0xc000289880, 0xc0001c4700)
/home/xxx/yyy/zzz/go/src/ppp/cws/pkg/websocket/docker.go:147 +0x83c fp=0xc000cc7bb8 sp=0xc000c>
net/http.HandlerFunc.ServeHTTP(...)
/usr/lib/golang/src/net/http/server.go:1995
ppp/cws/pkg/server.ValidateToken.func1(0xb45200, 0xc000289880, 0xc0001c4500)
/home/xxx/yyy/zzz/go/src/ppp/cws/pkg/server/auth.go:38 +0x293 fp=0xc000cc7c68 sp=0xc000cc7bb8 >
net/http.HandlerFunc.ServeHTTP(0xc00027a560, 0xb45200, 0xc000289880, 0xc0001c4500)
/usr/lib/golang/src/net/http/server.go:1995 +0x44 fp=0xc000cc7c90 sp=0xc000cc7c68 pc=0x72c654
net/http.(*ServeMux).ServeHTTP(0xf47d40, 0xb45200, 0xc000289880, 0xc0001c4500)
/usr/lib/golang/src/net/http/server.go:2375 +0x1d6 fp=0xc000cc7cf0 sp=0xc000cc7c90 pc=0x72e556
net/http.serverHandler.ServeHTTP(0xc000151380, 0xb45200, 0xc000289880, 0xc0001c4500)
/usr/lib/golang/src/net/http/server.go:2774 +0xa8 fp=0xc000cc7d20 sp=0xc000cc7cf0 pc=0x72f928
net/http.(*conn).serve(0xc0006a1400, 0xb46a00, 0xc0007767c0)
/usr/lib/golang/src/net/http/server.go:1878 +0x851 fp=0xc000cc7fc8 sp=0xc000cc7d20 pc=0x72b611
runtime.goexit()
/usr/lib/golang/src/runtime/asm_amd64.s:1337 +0x1 fp=0xc000cc7fd0 sp=0xc000cc7fc8 pc=0x45e4e1
created by net/http.(*Server).Serve
/usr/lib/golang/src/net/http/server.go:2884 +0x2f4
goroutine 1 [IO wait, 1 minutes]: