go-gun/gun/peer.go

package gun

import (
	"context"
	"fmt"
	"net/url"
	"sync"
	"time"
)

// ErrPeer is an error specific to a peer.
type ErrPeer struct {
	// Err is the error.
	Err error
	// Peer is the peer the error relates to.
	Peer *Peer
}

func (e *ErrPeer) Error() string { return fmt.Sprintf("Error on peer %v: %v", e.Peer, e.Err) }

// Peer is a known peer to Gun. It has a single connection. Some peers are
// "reconnectable" which means due to failure, they may be in a "bad" state
// awaiting reconnection.
type Peer struct {
	name       string
	newConn    func() (PeerConn, error)
	sleepOnErr time.Duration // TODO: would be better as backoff
	id         string

	connCurrent PeerConn
	connBad     bool // If true, don't try anything
	connLock    sync.Mutex
}

func newPeer(name string, newConn func() (PeerConn, error), sleepOnErr time.Duration) (*Peer, error) {
	p := &Peer{name: name, newConn: newConn, sleepOnErr: sleepOnErr}
	var err error
	if p.connCurrent, err = newConn(); err != nil {
		return nil, err
	}
	return p, nil
}

// ID is the identifier of the peer as given by the peer. It is empty if the
// peer wasn't asked for or didn't give an ID.
func (p *Peer) ID() string { return p.id }

// Name is the name of the peer which is usually the URL.
func (p *Peer) Name() string { return p.name }

// String is a string representation of the peer including whether it's
// connected.
func (p *Peer) String() string {
	id := ""
	if p.id != "" {
		id = "(id: " + p.id + ")"
	}
	connStatus := "disconnected"
	if conn := p.Conn(); conn != nil {
		connStatus = "connected to " + conn.RemoteURL()
	}
	return fmt.Sprintf("Peer%v %v (%v)", id, p.name, connStatus)
}

func (p *Peer) reconnectSupported() bool {
	return p.sleepOnErr > 0
}

func (p *Peer) reconnect() (err error) {
	if !p.reconnectSupported() {
		return fmt.Errorf("Reconnect not supported")
	}
	p.connLock.Lock()
	defer p.connLock.Unlock()
	if p.connCurrent == nil && p.connBad {
		p.connBad = false
		if p.connCurrent, err = p.newConn(); err != nil {
			p.connBad = true
			time.AfterFunc(p.sleepOnErr, func() { p.reconnect() })
		}
	}
	return
}

// Conn is the underlying PeerConn. This can be nil if the peer is currently
// "bad" or closed.
func (p *Peer) Conn() PeerConn {
	p.connLock.Lock()
	defer p.connLock.Unlock()
	return p.connCurrent
}

func (p *Peer) markConnErrored(conn PeerConn) {
	if !p.reconnectSupported() {
		p.Close()
		return
	}
	p.connLock.Lock()
	defer p.connLock.Unlock()
	if conn == p.connCurrent {
		p.connCurrent = nil
		p.connBad = true
		conn.Close()
		time.AfterFunc(p.sleepOnErr, func() { p.reconnect() })
	}
}

func (p *Peer) send(ctx context.Context, msg *Message, moreMsgs ...*Message) (ok bool, err error) {
	conn := p.Conn()
	if conn == nil {
		return false, nil
	}
	// Clone them with peer "to"
	updatedMsg := &Message{}
	*updatedMsg = *msg
	updatedMsg.To = conn.RemoteURL()
	updatedMoreMsgs := make([]*Message, len(moreMsgs))
	for i, moreMsg := range moreMsgs {
		updatedMoreMsg := &Message{}
		*updatedMoreMsg = *moreMsg
		updatedMoreMsg.To = conn.RemoteURL()
		updatedMoreMsgs[i] = updatedMoreMsg
	}
	if err = conn.Send(ctx, updatedMsg, updatedMoreMsgs...); err != nil {
		p.markConnErrored(conn)
		return false, err
	}
	return true, nil
}

func (p *Peer) receive(ctx context.Context) (ok bool, msgs []*Message, err error) {
	if conn := p.Conn(); conn == nil {
		return false, nil, nil
	} else if msgs, err = conn.Receive(ctx); err != nil {
		p.markConnErrored(conn)
		return false, nil, err
	} else {
		return true, msgs, nil
	}
}

// Close closes the peer and the connection is connected.
func (p *Peer) Close() error {
	p.connLock.Lock()
	defer p.connLock.Unlock()
	var err error
	if p.connCurrent != nil {
		err = p.connCurrent.Close()
		p.connCurrent = nil
	}
	p.connBad = false
	return err
}

// Closed is whether the peer is closed.
func (p *Peer) Closed() bool {
	p.connLock.Lock()
	defer p.connLock.Unlock()
	return p.connCurrent == nil && !p.connBad
}

// PeerConn is a single peer connection.
type PeerConn interface {
	// Send sends the given message (and maybe others) to the peer. The context
	// governs just this send.
	Send(ctx context.Context, msg *Message, moreMsgs ...*Message) error
	// Receive waits for the next message (or set of messages if sent at once)
	// from a peer. The context can be used to control a timeout.
	Receive(ctx context.Context) ([]*Message, error)
	// RemoteURL is the URL this peer is connected via.
	RemoteURL() string
	// Close closes this connection.
	Close() error
}

// PeerURLSchemes is the map that maps URL schemes to factory functions to
// create the connection. Currently "http" and "https" simply defer to "ws" and
// "wss" respectively. "ws" and "wss" use DialPeerConnWebSocket.
var PeerURLSchemes map[string]func(context.Context, *url.URL) (PeerConn, error)

func init() {
	PeerURLSchemes = map[string]func(context.Context, *url.URL) (PeerConn, error){
		"http": func(ctx context.Context, peerURL *url.URL) (PeerConn, error) {
			schemeChangedURL := &url.URL{}
			*schemeChangedURL = *peerURL
			schemeChangedURL.Scheme = "ws"
			return DialPeerConnWebSocket(ctx, schemeChangedURL)
		},
		"https": func(ctx context.Context, peerURL *url.URL) (PeerConn, error) {
			schemeChangedURL := &url.URL{}
			*schemeChangedURL = *peerURL
			schemeChangedURL.Scheme = "wss"
			return DialPeerConnWebSocket(ctx, schemeChangedURL)
		},
		"ws": func(ctx context.Context, peerURL *url.URL) (PeerConn, error) {
			return DialPeerConnWebSocket(ctx, peerURL)
		},
		"wss": func(ctx context.Context, peerURL *url.URL) (PeerConn, error) {
			return DialPeerConnWebSocket(ctx, peerURL)
		},
	}
}

// NewPeerConn connects to a peer for the given URL.
func NewPeerConn(ctx context.Context, peerURL string) (PeerConn, error) {
	if parsedURL, err := url.Parse(peerURL); err != nil {
		return nil, err
	} else if peerNew := PeerURLSchemes[parsedURL.Scheme]; peerNew == nil {
		return nil, fmt.Errorf("Unknown peer URL scheme %v", parsedURL.Scheme)
	} else {
		return peerNew(ctx, parsedURL)
	}
}