client interfaces compose better, remove some globals, test ConnectLoop

client/snowflake.go

@@ -5,6 +5,7 @@ import (
 	"bufio"
 	"errors"
 	"flag"
+	"fmt"
 	"io"
 	"log"
 	"net"
@@ -22,14 +23,12 @@ var ptInfo pt.ClientInfo
 
 const (
 	ReconnectTimeout  = 10
-	SnowflakeCapacity = 3
+	SnowflakeCapacity = 1
 )
 
 var brokerURL string
 var frontDomain string
 var iceServers IceServerList
-var snowflakeChan = make(chan *webRTCConn, 1)
-var broker *BrokerChannel
 
 // When a connection handler starts, +1 is written to this channel; when it
 // ends, -1 is written.
@@ -50,62 +49,110 @@ func copyLoop(a, b net.Conn) {
 	log.Println("copy loop ended")
 }
 
-// Interface that matches both webrtc.DataChannel and for testing.
+// Interface for catching Snowflakes.
+type Tongue interface {
+	Catch() (*webRTCConn, error)
+}
+
+// Interface for the Snowflake transport. (usually a webrtc.DataChannel)
 type SnowflakeChannel interface {
 	Send([]byte)
 	Close() error
 }
 
-// Maintain |SnowflakeCapacity| number of available WebRTC connections, to
-// transfer to the Tor SOCKS handler when needed.
-func SnowflakeConnectLoop() {
-	transport := CreateBrokerTransport()
-	broker = NewBrokerChannel(brokerURL, frontDomain, transport)
-	for {
-		numRemotes := len(webrtcRemotes)
-		if numRemotes >= SnowflakeCapacity {
-			log.Println("At Capacity: ", numRemotes, "snowflake. Re-checking in 10s")
-			<-time.After(time.Second * 10)
-			continue
-		}
-		s, err := dialWebRTC()
-		if nil == s || nil != err {
-			log.Println("WebRTC Error: ", err, " retrying...")
-			<-time.After(time.Second * ReconnectTimeout)
-			continue
-		}
-		snowflakeChan <- s
-	}
+// Collect and track available remote WebRTC Peers, to switch between if the
+// current one disconnects.
+// Right now, it is only possible to use one remote in a circuit. This can be
+// updated once multiplexed transport on a single circuit is available.
+type Peers struct {
+	Tongue
+
+	snowflakeChan chan *webRTCConn
+	current       *webRTCConn
+	capacity      int
+	maxedChan     chan struct{}
 }
 
-// Initialize a WebRTC Connection.
-func dialWebRTC() (*webRTCConn, error) {
-	// TODO: [#3] Fetch ICE server information from Broker.
-	// TODO: [#18] Consider TURN servers here too.
-	config := webrtc.NewConfiguration(iceServers...)
-	if nil == broker {
-		return nil, errors.New("Failed to prepare BrokerChannel")
-	}
-	connection := NewWebRTCConnection(config, broker)
-	err := connection.Connect()
-	return connection, err
+func NewPeers(max int) *Peers {
+	p := &Peers{capacity: max}
+	p.snowflakeChan = make(chan *webRTCConn, max)
+	p.maxedChan = make(chan struct{}, 1)
+	return p
 }
 
-func endWebRTC() {
+// Find, connect, and add a new peer to the internal collection.
+func (p *Peers) FindSnowflake() (*webRTCConn, error) {
+	if p.Count() >= p.capacity {
+		s := fmt.Sprintf("At capacity [%d/%d]", p.Count(), p.capacity)
+		p.maxedChan <- struct{}{}
+		return nil, errors.New(s)
+	}
+	connection, err := p.Catch()
+	if err != nil {
+		return nil, err
+	}
+	return connection, nil
+}
+
+// TODO: Needs fixing.
+func (p *Peers) Count() int {
+	return len(p.snowflakeChan)
+}
+
+// Close all remote peers.
+func (p *Peers) End() {
 	log.Printf("WebRTC: interruped")
-	for _, r := range webrtcRemotes {
+	if nil != p.current {
+		p.current.Close()
+	}
+	for r := range p.snowflakeChan {
 		r.Close()
 	}
 }
 
+// Maintain |SnowflakeCapacity| number of available WebRTC connections, to
+// transfer to the Tor SOCKS handler when needed.
+func ConnectLoop(peers *Peers) {
+	for {
+		s, err := peers.FindSnowflake()
+		if nil == s || nil != err {
+			log.Println("WebRTC Error:", err,
+				" Retrying in", ReconnectTimeout, "seconds...")
+			<-time.After(time.Second * ReconnectTimeout)
+			continue
+		}
+		peers.snowflakeChan <- s
+		<-time.After(time.Second)
+	}
+}
+
+// Implements |Tongue|
+type WebRTCDialer struct {
+	*BrokerChannel
+}
+
+// Initialize a WebRTC Connection by signaling through the broker.
+func (w WebRTCDialer) Catch() (*webRTCConn, error) {
+	if nil == w.BrokerChannel {
+		return nil, errors.New("Cannot Dial WebRTC without a BrokerChannel.")
+	}
+	// TODO: [#3] Fetch ICE server information from Broker.
+	// TODO: [#18] Consider TURN servers here too.
+	config := webrtc.NewConfiguration(iceServers...)
+	connection := NewWebRTCConnection(config, w.BrokerChannel)
+	err := connection.Connect()
+	return connection, err
+}
+
 // Establish a WebRTC channel for SOCKS connections.
-func handler(conn *pt.SocksConn) error {
+func handler(conn *pt.SocksConn, peers *Peers) error {
 	handlerChan <- 1
 	defer func() {
 		handlerChan <- -1
 	}()
 	// Wait for an available WebRTC remote...
-	remote, ok := <-snowflakeChan
+	remote, ok := <-peers.snowflakeChan
+	peers.current = remote
 	if remote == nil || !ok {
 		conn.Reject()
 		return errors.New("handler: Received invalid Snowflake")
@@ -125,7 +172,7 @@ func handler(conn *pt.SocksConn) error {
 	return nil
 }
 
-func acceptLoop(ln *pt.SocksListener) error {
+func acceptLoop(ln *pt.SocksListener, peers *Peers) error {
 	defer ln.Close()
 	for {
 		log.Println("SOCKS listening...", ln)
@@ -138,7 +185,7 @@ func acceptLoop(ln *pt.SocksListener) error {
 			return err
 		}
 		go func() {
-			err := handler(conn)
+			err := handler(conn, peers)
 			if err != nil {
 				log.Printf("handler error: %s", err)
 			}
@@ -146,6 +193,7 @@ func acceptLoop(ln *pt.SocksListener) error {
 	}
 }
 
+// TODO: Fix since multiplexing changes access to remotes.
 func readSignalingMessages(f *os.File) {
 	log.Printf("readSignalingMessages")
 	s := bufio.NewScanner(f)
@@ -157,10 +205,10 @@ func readSignalingMessages(f *os.File) {
 			log.Printf("ignoring invalid signal message %+q", msg)
 			continue
 		}
-		webrtcRemotes[0].answerChannel <- sdp
+		// webrtcRemotes[0].answerChannel <- sdp
 	}
 	log.Printf("close answerChannel")
-	close(webrtcRemotes[0].answerChannel)
+	// close(webrtcRemotes[0].answerChannel)
 	if err := s.Err(); err != nil {
 		log.Printf("signal FIFO: %s", err)
 	}
@@ -204,8 +252,13 @@ func main() {
 		go readSignalingMessages(signalFile)
 	}
 
-	webrtcRemotes = make(map[int]*webRTCConn)
-	go SnowflakeConnectLoop()
+	// Prepare WebRTC Peers and the Broker, then accumulate connections.
+	// TODO: Expose remote peer capacity as a flag?
+	remotes := NewPeers(SnowflakeCapacity)
+	broker := NewBrokerChannel(brokerURL, frontDomain, CreateBrokerTransport())
+
+	remotes.Tongue = WebRTCDialer{broker}
+	go ConnectLoop(remotes)
 
 	ptInfo, err = pt.ClientSetup(nil)
 	if err != nil {
@@ -221,12 +274,13 @@ func main() {
 	for _, methodName := range ptInfo.MethodNames {
 		switch methodName {
 		case "snowflake":
+			// TODO: Be able to recover when SOCKS dies.
 			ln, err := pt.ListenSocks("tcp", "127.0.0.1:0")
 			if err != nil {
 				pt.CmethodError(methodName, err.Error())
 				break
 			}
-			go acceptLoop(ln)
+			go acceptLoop(ln, remotes)
 			pt.Cmethod(methodName, ln.Version(), ln.Addr())
 			listeners = append(listeners, ln)
 		default:
@@ -234,7 +288,6 @@ func main() {
 		}
 	}
 	pt.CmethodsDone()
-	defer endWebRTC()
 
 	var numHandlers int = 0
 	var sig os.Signal
@@ -254,6 +307,8 @@ func main() {
 		ln.Close()
 	}
 
+	remotes.End()
+
 	// wait for second signal or no more handlers
 	sig = nil
 	for sig == nil && numHandlers != 0 {