cwtch/peer/connections/peerpeerconnection.go

package connections

import (
	"cwtch.im/cwtch/model"
	"cwtch.im/cwtch/peer/peer"
	"cwtch.im/cwtch/protocol"
	"git.openprivacy.ca/openprivacy/libricochet-go"
	"git.openprivacy.ca/openprivacy/libricochet-go/application"
	"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"
	"time"
)

// PeerPeerConnection encapsulates a single outgoing cwtchPeer->cwtchPeer connection
type PeerPeerConnection struct {
	connection.AutoConnectionHandler
	PeerHostname string
	state        ConnectionState
	connection   *connection.Connection
	profile      *model.Profile
	dataHandler  func(string, []byte) []byte
	aif          application.ApplicationInstanceFactory
	acn          connectivity.ACN
}

// NewPeerPeerConnection creates a new peer connection for the given hostname and profile.
func NewPeerPeerConnection(acn connectivity.ACN, peerhostname string, profile *model.Profile, dataHandler func(string, []byte) []byte, aif application.ApplicationInstanceFactory) *PeerPeerConnection {
	ppc := new(PeerPeerConnection)
	ppc.acn = acn
	ppc.PeerHostname = peerhostname
	ppc.profile = profile
	ppc.dataHandler = dataHandler
	ppc.aif = aif
	ppc.Init()
	return ppc
}

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

// HandleGroupInvite passes the given group invite tothe profile
func (ppc *PeerPeerConnection) HandleGroupInvite(gci *protocol.GroupChatInvite) {
	ppc.profile.ProcessInvite(gci, ppc.PeerHostname)
}

// HandlePacket handles data packets on the optional data channel
func (ppc *PeerPeerConnection) HandlePacket(data []byte) []byte {
	return ppc.dataHandler(ppc.PeerHostname, data)
}

// SendPacket sends data packets on the optional data channel
func (ppc *PeerPeerConnection) SendPacket(data []byte) {
	ppc.WaitTilAuthenticated()
	ppc.connection.Do(func() error {
		channel := ppc.connection.Channel("im.cwtch.peer.data", channels.Outbound)
		if channel != nil {
			peerchannel, ok := channel.Handler.(*peer.CwtchPeerDataChannel)
			if ok {
				log.Debugf("Sending packet\n")
				peerchannel.SendMessage(data)
			}
		}
		return nil
	})
}

// DoOnChannel performs an operation on the requested channel
func (ppc *PeerPeerConnection) DoOnChannel(ctype string, direction channels.Direction, doSomethingWith func(channel *channels.Channel)) {
	ppc.WaitTilAuthenticated()
	ppc.connection.Do(func() error {
		channel := ppc.connection.Channel(ctype, direction)
		if channel != nil {
			doSomethingWith(channel)
		}
		return nil
	})
}

// SendGroupInvite sends the given serialized invite packet to the Peer
func (ppc *PeerPeerConnection) SendGroupInvite(invite []byte) {
	ppc.WaitTilAuthenticated()
	ppc.connection.Do(func() error {
		channel := ppc.connection.Channel("im.cwtch.peer", channels.Outbound)
		if channel != nil {
			peerchannel, ok := channel.Handler.(*peer.CwtchPeerChannel)
			if ok {
				log.Debugf("Sending group invite packet\n")
				peerchannel.SendMessage(invite)
			}
		}
		return nil
	})
}

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

// Run manages the setup and teardown of a peer->peer connection
func (ppc *PeerPeerConnection) Run() error {
	ppc.state = CONNECTING
	rc, err := goricochet.Open(ppc.acn, ppc.PeerHostname)
	if err == nil {
		ppc.connection = rc
		ppc.state = CONNECTED
		_, err := connection.HandleOutboundConnection(ppc.connection).ProcessAuthAsV3Client(identity.InitializeV3(ppc.profile.Name, &ppc.profile.Ed25519PrivateKey, &ppc.profile.Ed25519PublicKey))
		if err == nil {
			ppc.state = AUTHENTICATED
			go func() {
				ppc.connection.Do(func() error {
					ppc.connection.RequestOpenChannel("im.cwtch.peer", &peer.CwtchPeerChannel{Handler: ppc})
					return nil
				})

				if ppc.dataHandler != nil {
					ppc.connection.Do(func() error {
						ppc.connection.RequestOpenChannel("im.cwtch.peer.data", &peer.CwtchPeerDataChannel{Handler: ppc})
						return nil
					})
				}

				handlers := ppc.aif.GetHandlers()
				for i := range handlers {
					ppc.connection.Do(func() error {
						ppc.connection.RequestOpenChannel(handlers[i], ppc.aif.GetHandler(handlers[i])(ppc.aif.GetApplicationInstance(ppc.connection))())
						return nil
					})
				}
			}()

			ppc.connection.Process(ppc)
		}
	}
	ppc.state = FAILED
	return err
}

// Close closes the connection
func (ppc *PeerPeerConnection) Close() {
	ppc.state = KILLED
	if ppc.connection != nil {
		ppc.connection.Conn.Close()
	}
}