groupme-lib/real_time.go

package groupme

import (
	"context"
	"errors"
	"gitea.watsonlabs.net/watsonb8/fayec"
	"gitea.watsonlabs.net/watsonb8/fayec/message"
	"gitea.watsonlabs.net/watsonb8/fayec/subscription"
	"log"
	"strings"
	"sync"
	"time"
)

const (
	PushServer   = "wss://push.groupme.com/faye"
	userChannel  = "/user/"
	groupChannel = "/group/"
	dmChannel    = "/direct_message/"
)

var (
	ErrHandlerNotFound    = errors.New("Handler not found")
	ErrListenerNotStarted = errors.New("GroupMe listener not started")
)

var concur = sync.Mutex{}

type HandlerAll interface {
	Handler

	//of self
	HandlerText
	HandlerLike
	HandlerMembership

	//of group
	HandleGroupTopic
	HandleGroupAvatar
	HandleGroupName
	HandleGroupLikeIcon

	//of group members
	HandleMemberNewNickname
	HandleMemberNewAvatar
	HandleMembers
}
type Handler interface {
	HandleError(error)
}
type HandlerText interface {
	HandleTextMessage(Message)
}
type HandlerLike interface {
	HandleLike(Message)
}
type HandlerMembership interface {
	HandleJoin(ID)
}

// Group Handlers
type HandleGroupTopic interface {
	HandleGroupTopic(group ID, newTopic string)
}

type HandleGroupName interface {
	HandleGroupName(group ID, newName string)
}
type HandleGroupAvatar interface {
	HandleGroupAvatar(group ID, newAvatar string)
}
type HandleGroupLikeIcon interface {
	HandleLikeIcon(group ID, PackID, PackIndex int, Type string)
}

// Group member handlers
type HandleMemberNewNickname interface {
	HandleNewNickname(group ID, user ID, newName string)
}

type HandleMemberNewAvatar interface {
	HandleNewAvatarInGroup(group ID, user ID, avatarURL string)
}
type HandleMembers interface {
	//HandleNewMembers returns only partial member with id and nickname; added is false if removing
	HandleMembers(group ID, members []Member, added bool)
}

// PushSubscription manages real time subscription
type PushSubscription struct {
	channel       chan message.Data
	client        *fayec.Client
	handlers      map[string][]Handler // key == token
	LastConnected int64
}

// NewPushSubscription creates and returns a push subscription object
func NewPushSubscription(context context.Context) PushSubscription {

	r := PushSubscription{
		channel:  make(chan message.Data),
		handlers: make(map[string][]Handler),
	}

	return r
}

func (r *PushSubscription) AddHandler(h Handler, authToken string) {
	if r.handlers[authToken] == nil {
		r.handlers[authToken] = []Handler{h}
	} else {
		r.handlers[authToken] = append(r.handlers[authToken], h)
	}
	//r.handlers = append(r.handlers, h)
}

// AddFullHandler is the same as AddHandler except it ensures the interface implements everything
func (r *PushSubscription) AddFullHandler(h HandlerAll, authToken string) {
	if r.handlers[authToken] == nil {
		r.handlers[authToken] = []Handler{h}
	} else {
		r.handlers[authToken] = append(r.handlers[authToken], h)
	}

	//r.handlers = append(r.handlers, h)
}

var RealTimeHandlers map[string]func(r *PushSubscription, channel string, authToken string, data ...interface{})
var RealTimeSystemHandlers map[string]func(r *PushSubscription, channel string, id ID, authToken string, rawData []byte)

// Listen connects to GroupMe. Runs in Goroutine.
func (r *PushSubscription) Connect(context context.Context) error {
	c, err := fayec.NewClient(PushServer)
	if err != nil {
		return err
	}
	r.client = c

	return nil
}

// SubscribeToUser to users
func (r *PushSubscription) SubscribeToUser(context context.Context, id ID, authToken string) error {
	return r.subscribeWithPrefix(userChannel, context, id, authToken)
}

// SubscribeToGroup to groups for typing notification
func (r *PushSubscription) SubscribeToGroup(context context.Context, id ID, authToken string) error {
	return r.subscribeWithPrefix(groupChannel, context, id, authToken)
}

// SubscribeToDM to users
func (r *PushSubscription) SubscribeToDM(context context.Context, id ID, authToken string) error {
	id = ID(strings.Replace(id.String(), "+", "_", 1))
	return r.subscribeWithPrefix(dmChannel, context, id, authToken)
}

func (r *PushSubscription) subscribeWithPrefix(prefix string, context context.Context, groupID ID, authToken string) error {
	concur.Lock()
	defer concur.Unlock()
	if r.client == nil {
		return ErrListenerNotStarted
	}

	var sub *subscription.Subscription
	sub, err := r.client.Subscribe(prefix+groupID.String(), authToken)
	if err != nil {
		panic(err)
	}

	err = sub.OnMessage(func(channel string, data message.Data) {
		r.LastConnected = time.Now().Unix()
		dataMap := data.(map[string]interface{})
		content := dataMap["subject"]
		contentType := dataMap["type"].(string)

		handler, ok := RealTimeHandlers[contentType]
		if !ok {
			if contentType == "ping" ||
				len(contentType) == 0 ||
				content == "" {
				return
			}
			log.Println("Unable to handle GroupMe message type", contentType)
		} else {
			handler(r, channel, authToken, content)
		}
	})

	return nil
}

// Connected check if connected
func (r *PushSubscription) Connected() bool {
	return r.LastConnected+30 >= time.Now().Unix()
}