cwtch/protocol/connections/peerapp.go

package connections

import (
	"cwtch.im/cwtch/event"
	"cwtch.im/tapir"
	"cwtch.im/tapir/applications"
	"encoding/json"
	"git.openprivacy.ca/openprivacy/log"
	"sync"
)

const cwtchCapability = tapir.Capability("cwtchCapability")

// PeerApp encapsulates the behaviour of a Cwtch Peer
type PeerApp struct {
	applications.AuthApp
	connection        tapir.Connection
	MessageHandler    func(string, string, string, []byte)
	RetValHandler     func(string, []byte, []byte)
	IsBlocked         func(string) bool
	IsAllowed         func(string) bool
	OnAcknowledgement func(string, string)
	OnAuth            func(string)
	OnClose           func(string)
	OnConnecting      func(string)

	getValRequests sync.Map // [string]string eventID:Data
}

// PeerMessage is an encapsulation that can be used by higher level applications
type PeerMessage struct {
	ID      string // A unique Message ID (primarily used for acknowledgments)
	Context string // A unique context identifier i.e. im.cwtch.chat
	Data    []byte // The serialized data packet.
}

type peerGetVal struct {
	Scope, Path string
}

type peerRetVal struct {
	Val    string
	Exists bool
}

// NewInstance should always return a new instantiation of the application.
func (pa *PeerApp) NewInstance() tapir.Application {
	newApp := new(PeerApp)
	newApp.MessageHandler = pa.MessageHandler
	newApp.IsBlocked = pa.IsBlocked
	newApp.IsAllowed = pa.IsAllowed
	newApp.OnAcknowledgement = pa.OnAcknowledgement
	newApp.OnAuth = pa.OnAuth
	newApp.OnClose = pa.OnClose
	newApp.OnConnecting = pa.OnConnecting
	newApp.RetValHandler = pa.RetValHandler
	return newApp
}

// Init is run when the connection is first started.
func (pa *PeerApp) Init(connection tapir.Connection) {
	// First run the Authentication App
	pa.AuthApp.Init(connection)

	if connection.HasCapability(applications.AuthCapability) {

		pa.connection = connection
		connection.SetCapability(cwtchCapability)

		if pa.IsBlocked(connection.Hostname()) {
			pa.connection.Close()
			pa.OnClose(connection.Hostname())
		} else {
			pa.OnAuth(connection.Hostname())
			go pa.listen()
		}
	} else {
		// The auth protocol wasn't completed, we can safely shutdown the connection
		connection.Close()
	}
}

func (pa *PeerApp) listen() {
	for {
		message := pa.connection.Expect()
		if len(message) == 0 {
			log.Debugf("0 byte read, socket has likely failed. Closing the listen goroutine")
			pa.OnClose(pa.connection.Hostname())
			return
		}
		var peerMessage PeerMessage
		err := json.Unmarshal(message, &peerMessage)
		if err == nil {
			switch peerMessage.Context {
			case event.ContextAck:
				pa.OnAcknowledgement(pa.connection.Hostname(), peerMessage.ID)
			case event.ContextRetVal:
				req, ok := pa.getValRequests.Load(peerMessage.ID)
				if ok {
					reqStr := []byte(req.(string))
					pa.RetValHandler(pa.connection.Hostname(), reqStr, peerMessage.Data)
					pa.getValRequests.Delete(peerMessage.ID)
				}
			default:
				if pa.IsAllowed(pa.connection.Hostname()) {
					pa.MessageHandler(pa.connection.Hostname(), peerMessage.ID, peerMessage.Context, peerMessage.Data)

					// Acknowledge the message
					pa.SendMessage(PeerMessage{peerMessage.ID, event.ContextAck, []byte{}})
				}
			}
		} else {
			log.Errorf("Error unmarshalling PeerMessage package: %x %v", message, err)
		}
	}
}

// SendMessage sends the peer a preformatted message
// NOTE: This is a stub, we will likely want to extend this to better reflect the desired protocol
func (pa *PeerApp) SendMessage(message PeerMessage) {
	if message.Context == event.ContextGetVal {
		pa.getValRequests.Store(message.ID, string(message.Data))
	}
	serialized, _ := json.Marshal(message)
	pa.connection.Send(serialized)
}