cwtch/peer/connections/peerserverconnection.go

package connections

import (
	"crypto/rand"
	"cwtch.im/cwtch/peer/fetch"
	"cwtch.im/cwtch/peer/listen"
	"cwtch.im/cwtch/peer/send"
	"cwtch.im/cwtch/protocol"
	"errors"
	"git.openprivacy.ca/openprivacy/libricochet-go"
	"git.openprivacy.ca/openprivacy/libricochet-go/channels"
	"git.openprivacy.ca/openprivacy/libricochet-go/connection"
	"git.openprivacy.ca/openprivacy/libricochet-go/connectivity"
	"git.openprivacy.ca/openprivacy/libricochet-go/identity"
	"git.openprivacy.ca/openprivacy/libricochet-go/log"
	"golang.org/x/crypto/ed25519"
	"time"
)

// PeerServerConnection encapsulates a single Peer->Server connection
type PeerServerConnection struct {
	connection.AutoConnectionHandler
	Server     string
	state      ConnectionState
	connection *connection.Connection
	acn        connectivity.ACN

	GroupMessageHandler func(string, *protocol.GroupMessage)
}

// NewPeerServerConnection creates a new Peer->Server outbound connection
func NewPeerServerConnection(acn connectivity.ACN, serverhostname string) *PeerServerConnection {
	psc := new(PeerServerConnection)
	psc.acn = acn
	psc.Server = serverhostname
	psc.Init()
	return psc
}

// GetState returns the current connection state
func (psc *PeerServerConnection) GetState() ConnectionState {
	return psc.state
}

// WaitTilAuthenticated waits until the underlying connection is authenticated
func (psc *PeerServerConnection) WaitTilAuthenticated() {
	for {
		if psc.GetState() == AUTHENTICATED {
			break
		}
		time.Sleep(time.Second * 1)
	}
}

// Run manages the setup and teardown of a peer server connection
func (psc *PeerServerConnection) Run() error {
	log.Infof("Connecting to %v", psc.Server)
	rc, err := goricochet.Open(psc.acn, psc.Server)
	if err == nil {
		psc.connection = rc
		psc.state = CONNECTED
		pub, priv, err := ed25519.GenerateKey(rand.Reader)
		if err == nil {
			_, err := connection.HandleOutboundConnection(psc.connection).ProcessAuthAsV3Client(identity.InitializeV3("cwtchpeer", &priv, &pub))
			if err == nil {
				psc.state = AUTHENTICATED

				go func() {
					psc.connection.Do(func() error {
						psc.connection.RequestOpenChannel("im.cwtch.server.fetch", &fetch.CwtchPeerFetchChannel{Handler: psc})
						return nil
					})

					psc.connection.Do(func() error {
						psc.connection.RequestOpenChannel("im.cwtch.server.listen", &listen.CwtchPeerListenChannel{Handler: psc})
						return nil
					})
				}()
				psc.connection.Process(psc)
			}
		}
	}
	psc.state = FAILED
	return err
}

// Break makes Run() return and prevents processing, but doesn't close the connection.
func (psc *PeerServerConnection) Break() error {
	return psc.connection.Break()
}

// SendGroupMessage sends the given protocol message to the Server.
func (psc *PeerServerConnection) SendGroupMessage(gm *protocol.GroupMessage) error {
	if psc.state != AUTHENTICATED {
		return errors.New("peer is not yet connected & authenticated to server cannot send message")
	}

	err := psc.connection.Do(func() error {
		psc.connection.RequestOpenChannel("im.cwtch.server.send", &send.CwtchPeerSendChannel{})
		return nil
	})

	errCount := 0
	for errCount < 5 {
		time.Sleep(time.Second * 1)
		err = psc.connection.Do(func() error {
			channel := psc.connection.Channel("im.cwtch.server.send", channels.Outbound)
			if channel == nil {
				return errors.New("no channel found")
			}
			sendchannel, ok := channel.Handler.(*send.CwtchPeerSendChannel)
			if ok {
				return sendchannel.SendGroupMessage(gm)
			}
			return errors.New("channel is not a peer send channel (this should definitely not happen)")
		})

		if err != nil {
			errCount++
		} else {
			return nil
		}
	}

	return err
}

// Close shuts down the connection (freeing the handler goroutines)
func (psc *PeerServerConnection) Close() {
	psc.state = KILLED
	if psc.connection != nil {
		psc.connection.Conn.Close()
	}
}

// HandleGroupMessage passes the given group message back to the profile.
func (psc *PeerServerConnection) HandleGroupMessage(gm *protocol.GroupMessage) {
	log.Debugf("Received Group Message")
	psc.GroupMessageHandler(psc.Server, gm)
}