go-wsqueue
Introduction
- Golang API over gorilla/mux and gorilla/websocket
- Kind of AMQP but over websockets
Publish & Subscribe Pattern
Publish and subscribe semantics are implemented by Topics.
When you publish a message it goes to all the subscribers who are interested - so zero to many subscribers will receive a copy of the message. Only subscribers who had an active subscription at the time the broker receives the message will get a copy of the message.
Work queues Pattern
The main idea behind Work Queues (aka: Task Queues) is to avoid doing a resource-intensive task immediately and having to wait for it to complete. Instead we schedule the task to be done later. We encapsulate a task as a message and send it to the queue. A worker process running in the background will pop the tasks and eventually execute the job. When you run many workers the tasks will be shared between them.
Queues implement load balancer semantics. A single message will be received by exactly one consumer. If there are no consumers available at the time the message is sent it will be kept until a consumer is available that can process the message. If a consumer receives a message and does not acknowledge it before closing then the message will be redelivered to another consumer. A queue can have many consumers with messages load balanced across the available consumers.
Getting started
$ go get github.com/fsamin/go-wsqueue
Publish & Subscribe
Run this with 3 terminals :
go run main.go -server
go run main.go -client1
go run main.go -client2
main.go :
package main
import (
"flag"
"log"
"net/http"
"time"
"github.com/fsamin/go-wsqueue"
"github.com/gorilla/mux"
"github.com/jmcvetta/randutil"
)
var fServer = flag.Bool("server", false, "Run server")
var fClient1 = flag.Bool("client1", false, "Run client #1")
var fClient2 = flag.Bool("client2", false, "Run client #2")
func main() {
flag.Parse()
forever := make(chan bool)
if *fServer {
server()
}
if *fClient1 {
client("1")
}
if *fClient2 {
client("2")
}
<-forever
}
func server() {
r := mux.NewRouter()
s := wsqueue.NewServer(r, "")
q := s.CreateTopic("topic1")
q.OpenedConnectionHandler = func(c *wsqueue.Conn) {
log.Println("Welcome " + c.ID)
q.Publish("Welcome " + c.ID)
}
q.ClosedConnectionHandler = func(c *wsqueue.Conn) {
log.Println("Bye bye " + c.ID)
}
http.Handle("/", r)
go http.ListenAndServe("0.0.0.0:9000", r)
//Start send message to queue
go func() {
for {
time.Sleep(5 * time.Second)
s, _ := randutil.AlphaString(10)
q.Publish("message from goroutine 1 : " + s)
}
}()
go func() {
for {
time.Sleep(10 * time.Second)
s, _ := randutil.AlphaString(10)
q.Publish("message from goroutine 2 : " + s)
}
}()
}
func client(ID string) {
//Connect a client
go func() {
c := &wsqueue.Client{
Protocol: "ws",
Host: "localhost:9000",
Route: "/",
}
cMessage, cError, err := c.Subscribe("topic1")
if err != nil {
panic(err)
}
for {
select {
case m := <-cMessage:
log.Println("\n\n********* Client " + ID + " *********" + m.String() + "\n******************")
case e := <-cError:
log.Println("\n\n********* Client " + ID + " *********" + e.Error() + "\n******************")
}
}
}()
}
Load balanced Work Queues
Run this with 3 terminals :
go run main.go -server
go run main.go -client1
go run main.go -client2
main.go :
package main
import (
"flag"
"log"
"net/http"
"time"
"github.com/fsamin/go-wsqueue"
"github.com/gorilla/mux"
"github.com/jmcvetta/randutil"
)
var fServer = flag.Bool("server", false, "Run server")
var fClient1 = flag.Bool("client1", false, "Run client #1")
var fClient2 = flag.Bool("client2", false, "Run client #2")
func main() {
flag.Parse()
forever := make(chan bool)
if *fServer {
server()
}
if *fClient1 {
client("1")
}
if *fClient2 {
client("2")
}
<-forever
}
func server() {
r := mux.NewRouter()
s := wsqueue.NewServer(r, "")
q := s.CreateQueue("queue1", 10)
http.Handle("/", r)
go http.ListenAndServe("0.0.0.0:9000", r)
//Start send message to queue
go func() {
for {
time.Sleep(5 * time.Second)
s, _ := randutil.AlphaString(10)
q.Send("message from goroutine 1 : " + s)
}
}()
go func() {
for {
time.Sleep(6 * time.Second)
s, _ := randutil.AlphaString(10)
q.Send("message from goroutine 2 : " + s)
}
}()
}
func client(ID string) {
//Connect a client
go func() {
c := &wsqueue.Client{
Protocol: "ws",
Host: "localhost:9000",
Route: "/",
}
cMessage, cError, err := c.Listen("queue1")
if err != nil {
panic(err)
}
for {
select {
case m := <-cMessage:
log.Println("\n\n********* Client " + ID + " *********" + m.String() + "\n******************")
case e := <-cError:
log.Println("\n\n********* Client " + ID + " *********" + e.Error() + "\n******************")
}
}
}()
}
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent