connect/server/redis_exchange.go

package server

import (
	"context"
	"encoding"
	"encoding/json"
	"fmt"

	"github.com/redis/go-redis/v9"
	"sikey.com/websocket/utils/zlog"
)

type Exchange struct {
	ctx      context.Context
	rdb      redis.UniversalClient
	serverId string

	Connect    chan *EventConnect
	Disconnect chan *EventDisconnect
	Message    chan *Message
}

func NewExchange(serverId string, rdb redis.UniversalClient) *Exchange {
	exc := &Exchange{
		ctx:      context.Background(),
		rdb:      rdb,
		serverId: serverId,

		Connect:    make(chan *EventConnect),
		Disconnect: make(chan *EventDisconnect),
		Message:    make(chan *Message),
	}

	go exc.remotelyEvent()
	return exc
}

var (
	channelEvent           = "client.event.*"
	connectChannelEvent    = "client.event.connect"
	disconnectChannelEvent = "client.event.disconnect"
	messageChannelEvent    = "client.event.message"

	clients = "client.users.%s"
)

type (
	EventConnect struct {
		UserId   string
		ServerId string
	}

	EventDisconnect struct {
		UserId   string
		ServerId string
	}
)

var _ encoding.BinaryMarshaler = (*EventConnect)(nil)
var _ encoding.BinaryUnmarshaler = (*EventConnect)(nil)

var _ encoding.BinaryMarshaler = (*EventDisconnect)(nil)
var _ encoding.BinaryUnmarshaler = (*EventDisconnect)(nil)

func (e *EventDisconnect) UnmarshalBinary(data []byte) error {
	return json.Unmarshal(data, e)
}

func (e *EventDisconnect) MarshalBinary() (data []byte, err error) {
	return json.Marshal(e)
}

func (e *EventConnect) UnmarshalBinary(data []byte) error {
	return json.Unmarshal(data, e)
}

func (e *EventConnect) MarshalBinary() (data []byte, err error) {
	return json.Marshal(e)
}

func (exc *Exchange) OnPublishConnect(client *Client) error {
	// Create clients list
	var err error
	err = exc.rdb.Set(exc.ctx, fmt.Sprintf(clients, client.UserId), exc.serverId, 0).Err()
	if err != nil {
		zlog.Error(err)
	}
	event := &EventConnect{ServerId: exc.serverId, UserId: client.UserId}
	err = exc.rdb.Publish(exc.ctx, connectChannelEvent, event).Err()
	if err != nil {
		zlog.Error(err)
	}
	return err
}

func (exc *Exchange) OnPublishDisconnect(client *Client) error {
	exc.rdb.Del(exc.ctx, fmt.Sprintf(clients, client.UserId), exc.serverId)
	event := &EventDisconnect{ServerId: exc.serverId, UserId: client.UserId}
	return exc.rdb.Publish(exc.ctx, disconnectChannelEvent, event).Err()
}

func (exc *Exchange) OnPublishMessage(client *Client, message *Message) error {
	err := exc.rdb.Publish(exc.ctx, messageChannelEvent, message).Err()
	if err != nil {
		zlog.Error(err)
		return err
	}
	return nil
}

func (exc *Exchange) remotelyEvent() {
	ctx := context.TODO()
	pubsub := exc.rdb.PSubscribe(ctx, channelEvent)
	defer pubsub.Close()

	for {
		rMsg, err := pubsub.ReceiveMessage(ctx)
		if err != nil {
			zlog.Error(err)
		}

		switch rMsg.Channel {
		case connectChannelEvent:
			var event = deserializeEventConnect([]byte(rMsg.Payload))
			if event.ServerId != exc.serverId {
				exc.Connect <- &EventConnect{UserId: event.UserId}
			}
		case disconnectChannelEvent:
			var event = deserializeEventDisconnect([]byte(rMsg.Payload))
			if event.ServerId != exc.serverId {
				exc.Disconnect <- &EventDisconnect{UserId: event.UserId}
			}
		case messageChannelEvent:
			exc.Message <- deserializeMessage([]byte(rMsg.Payload))
		}
	}
}

func deserializeEventConnect(bytes []byte) *EventConnect {
	var event EventConnect
	_ = json.Unmarshal(bytes, &event)
	return &event
}

func deserializeEventDisconnect(bytes []byte) *EventDisconnect {
	var event EventDisconnect
	_ = json.Unmarshal(bytes, &event)
	return &event
}