|
|
|
|
@ -2,12 +2,19 @@ package ws
|
|
|
|
|
|
|
|
|
|
import (
|
|
|
|
|
"encoding/json"
|
|
|
|
|
"fmt"
|
|
|
|
|
"log"
|
|
|
|
|
"net/http"
|
|
|
|
|
"strings"
|
|
|
|
|
"time"
|
|
|
|
|
|
|
|
|
|
"git.sp4ke.com/sp4ke/bit4sat/bus"
|
|
|
|
|
"git.sp4ke.com/sp4ke/bit4sat/db"
|
|
|
|
|
"github.com/gin-contrib/sessions"
|
|
|
|
|
"github.com/gin-gonic/gin"
|
|
|
|
|
"github.com/gorilla/websocket"
|
|
|
|
|
"github.com/mediocregopher/radix/v3"
|
|
|
|
|
"github.com/segmentio/ksuid"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
const (
|
|
|
|
|
@ -23,6 +30,8 @@ const (
|
|
|
|
|
|
|
|
|
|
// Maximum message size
|
|
|
|
|
maxMessageSize = 512
|
|
|
|
|
|
|
|
|
|
WebsocketIdName = "websocket-id"
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
var upgrader = websocket.Upgrader{
|
|
|
|
|
@ -33,22 +42,38 @@ var upgrader = websocket.Upgrader{
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
var C *Client
|
|
|
|
|
// Keep reference to clients so they can be used in goroutines
|
|
|
|
|
var C = make(map[*Client]bool)
|
|
|
|
|
|
|
|
|
|
// Interface to talk to websocket
|
|
|
|
|
type Client struct {
|
|
|
|
|
id ksuid.KSUID
|
|
|
|
|
|
|
|
|
|
// websocket connection
|
|
|
|
|
conn *websocket.Conn
|
|
|
|
|
|
|
|
|
|
// Buffered channel of outbound messages
|
|
|
|
|
send chan []byte
|
|
|
|
|
// PubSub message channel for this websocket
|
|
|
|
|
subChannel chan radix.PubSubMessage
|
|
|
|
|
|
|
|
|
|
// Name of main subscribed channel for this websocket
|
|
|
|
|
channelName string
|
|
|
|
|
|
|
|
|
|
// Listen to updates from redis
|
|
|
|
|
// upload ids registered to this client
|
|
|
|
|
uploadId string
|
|
|
|
|
|
|
|
|
|
// upload notifications channel
|
|
|
|
|
uploadNotifChannel chan radix.PubSubMessage
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
func (c *Client) readPump() {
|
|
|
|
|
defer c.conn.Close()
|
|
|
|
|
defer func() {
|
|
|
|
|
// Delete client from referenc list
|
|
|
|
|
delete(C, c)
|
|
|
|
|
|
|
|
|
|
// Close ws connection
|
|
|
|
|
c.conn.Close()
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
|
|
c.conn.SetReadLimit(maxMessageSize)
|
|
|
|
|
c.conn.SetReadDeadline(time.Now().Add(pongWait))
|
|
|
|
|
c.conn.SetPongHandler(func(string) error {
|
|
|
|
|
@ -82,10 +107,101 @@ func (c *Client) writePump() {
|
|
|
|
|
c.conn.Close()
|
|
|
|
|
}()
|
|
|
|
|
|
|
|
|
|
// Subscribe to upload notifications
|
|
|
|
|
// Subscribe to general notifications for this client
|
|
|
|
|
log.Printf("socket subscribing to %s", c.channelName)
|
|
|
|
|
if err := db.DB.RedisPubSub.Subscribe(c.subChannel,
|
|
|
|
|
c.channelName); err != nil {
|
|
|
|
|
log.Println(err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// subscribe to notifications related to this channel's registered upload
|
|
|
|
|
// ids
|
|
|
|
|
if err := db.DB.RedisPubSub.PSubscribe(c.uploadNotifChannel,
|
|
|
|
|
fmt.Sprintf("%s_*", bus.UploadUpdateChannelPrefix)); err != nil {
|
|
|
|
|
log.Println(err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
for {
|
|
|
|
|
select {
|
|
|
|
|
case msg := <-c.subChannel:
|
|
|
|
|
log.Printf("received msg %s on socket main channel", msg)
|
|
|
|
|
jsonMsg := bus.Message{}
|
|
|
|
|
if err := json.Unmarshal(msg.Message, &jsonMsg); err != nil {
|
|
|
|
|
log.Printf("ws error reading from pubsub: %s", err)
|
|
|
|
|
continue
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if jsonMsg.Type == bus.SetUploadId {
|
|
|
|
|
c.uploadId = jsonMsg.UploadId
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
case msg := <-c.uploadNotifChannel:
|
|
|
|
|
log.Printf("websocket received upload paid notification on channel %s", msg.Channel)
|
|
|
|
|
log.Printf("our registered ids are %s", c.uploadId)
|
|
|
|
|
|
|
|
|
|
// If we have no upload id registered for this client break
|
|
|
|
|
if c.uploadId == "" {
|
|
|
|
|
continue
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Check if the message matches our upload id
|
|
|
|
|
slice := strings.SplitN(msg.Channel, "_", 3)
|
|
|
|
|
if len(slice) != 3 {
|
|
|
|
|
log.Printf("error decoding channel name %s", msg.Channel)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
msgTarget := slice[2]
|
|
|
|
|
|
|
|
|
|
// If the target is any of our registered upload ids
|
|
|
|
|
// broadcast the message
|
|
|
|
|
|
|
|
|
|
if c.uploadId == msgTarget {
|
|
|
|
|
log.Println("this message is for us sending to client")
|
|
|
|
|
|
|
|
|
|
jsonMsg := bus.Message{}
|
|
|
|
|
err := json.Unmarshal(msg.Message, &jsonMsg)
|
|
|
|
|
if err != nil {
|
|
|
|
|
log.Printf("unmarshal error %s", err)
|
|
|
|
|
break
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
switch jsonMsg.Type {
|
|
|
|
|
|
|
|
|
|
case bus.PaymentReceived:
|
|
|
|
|
|
|
|
|
|
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
|
|
|
|
|
|
|
|
|
|
w, err := c.conn.NextWriter(websocket.TextMessage)
|
|
|
|
|
if err != nil {
|
|
|
|
|
log.Println(err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
socketMsg := ProtoMessage{}
|
|
|
|
|
socketMsg.Type = messageTypes[invoicePaid]
|
|
|
|
|
socketMsg.Data = jsonMsg.Data
|
|
|
|
|
|
|
|
|
|
enc := json.NewEncoder(w)
|
|
|
|
|
err = enc.Encode(socketMsg)
|
|
|
|
|
if err != nil {
|
|
|
|
|
log.Println(err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// No need to encode the message it's already in json
|
|
|
|
|
if err = w.Close(); err != nil {
|
|
|
|
|
fmt.Println(err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Handle different message types
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
//case <-testTicker.C:
|
|
|
|
|
//c.conn.SetWriteDeadline(time.Now().Add(writeWait))
|
|
|
|
|
|
|
|
|
|
@ -117,10 +233,9 @@ func (c *Client) writePump() {
|
|
|
|
|
|
|
|
|
|
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
|
|
|
|
|
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
|
|
|
|
|
log.Println(err)
|
|
|
|
|
log.Printf("websocket ping error: %s", err)
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
@ -128,7 +243,24 @@ func (c *Client) writePump() {
|
|
|
|
|
func Serve(c *gin.Context) {
|
|
|
|
|
log.Println("websocket request")
|
|
|
|
|
|
|
|
|
|
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
|
|
|
|
|
session := sessions.Default(c)
|
|
|
|
|
|
|
|
|
|
// Create unique id for this session in order to store it
|
|
|
|
|
// in the websocket bus
|
|
|
|
|
|
|
|
|
|
socketSessionId, err := ksuid.NewRandomWithTime(time.Now())
|
|
|
|
|
if err != nil {
|
|
|
|
|
log.Println("error generating ksuid for websocket")
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
log.Printf("using socket id: %s", socketSessionId)
|
|
|
|
|
|
|
|
|
|
session.Set(WebsocketIdName, socketSessionId.String())
|
|
|
|
|
session.Save()
|
|
|
|
|
|
|
|
|
|
log.Printf("Writing socket session id to header: %s", c.Writer.Header())
|
|
|
|
|
|
|
|
|
|
conn, err := upgrader.Upgrade(c.Writer, c.Request, c.Writer.Header())
|
|
|
|
|
if err != nil {
|
|
|
|
|
if _, ok := err.(websocket.HandshakeError); !ok {
|
|
|
|
|
log.Printf("handshake error: %s", err)
|
|
|
|
|
@ -138,8 +270,16 @@ func Serve(c *gin.Context) {
|
|
|
|
|
return
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
C = &Client{conn: conn, send: make(chan []byte, 256)}
|
|
|
|
|
client := &Client{
|
|
|
|
|
id: socketSessionId,
|
|
|
|
|
conn: conn,
|
|
|
|
|
subChannel: make(chan radix.PubSubMessage),
|
|
|
|
|
uploadNotifChannel: make(chan radix.PubSubMessage),
|
|
|
|
|
channelName: fmt.Sprintf("%s_%s", bus.WebsocketPubSubPrefix, socketSessionId),
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
C[client] = true
|
|
|
|
|
|
|
|
|
|
go C.writePump()
|
|
|
|
|
go C.readPump()
|
|
|
|
|
go client.writePump()
|
|
|
|
|
go client.readPump()
|
|
|
|
|
}
|
|
|
|
|
|
Chakib Benziane
5 years ago