[tor-commits] [snowflake/master] client interfaces compose better, remove some globals, test ConnectLoop

20 May 2016

commit 6b8568cc6cf7186d86e844a6f1460c9af802575a
Author: Serene Han keroserene+git@gmail.com
Date:   Thu May 19 18:06:34 2016 -0700
client interfaces compose better, remove some globals, test ConnectLoop
---
 client/client_test.go |  85 +++++++++++++++++++++++++-------
 client/snowflake.go   | 133 +++++++++++++++++++++++++++++++++++---------------
 client/webrtc.go      |  43 +++++++++-------
 3 files changed, 186 insertions(+), 75 deletions(-)

diff --git a/client/client_test.go b/client/client_test.go
index 41dc5e6..93e0422 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -2,6 +2,7 @@ package main
import (
    "bytes"
+	"fmt"
    "github.com/keroserene/go-webrtc"
    . "github.com/smartystreets/goconvey/convey"
    "io/ioutil"
@@ -48,9 +49,64 @@ func (m *MockTransport) RoundTrip(req *http.Request) (*http.Response, error) {
    return r, nil
 }
-func TestConnect(t *testing.T) {
+type FakeDialer struct{}
+
+func (w FakeDialer) Catch() (*webRTCConn, error) {
+	fmt.Println("Caught a dummy snowflake.")
+	return &webRTCConn{}, nil
+}
+
+func TestSnowflakeClient(t *testing.T) {
    Convey("Snowflake", t, func() {
-		webrtcRemotes = make(map[int]*webRTCConn)
+
+		Convey("Peers", func() {
+
+			Convey("WebRTC ConnectLoop continues until capacity of 1.\n", func() {
+				peers := NewPeers(1)
+				peers.Tongue = FakeDialer{}
+
+				go ConnectLoop(peers)
+				<-peers.maxedChan
+
+				So(peers.Count(), ShouldEqual, 1)
+				r := <-peers.snowflakeChan
+				So(r, ShouldNotBeNil)
+				So(peers.Count(), ShouldEqual, 0)
+			})
+
+			Convey("WebRTC ConnectLoop continues until capacity of 3.\n", func() {
+				peers := NewPeers(3)
+				peers.Tongue = FakeDialer{}
+
+				go ConnectLoop(peers)
+				<-peers.maxedChan
+				So(peers.Count(), ShouldEqual, 3)
+				<-peers.snowflakeChan
+				<-peers.snowflakeChan
+				<-peers.snowflakeChan
+				So(peers.Count(), ShouldEqual, 0)
+			})
+
+			Convey("WebRTC ConnectLoop continues filling when Snowflakes disconnect.\n", func() {
+				peers := NewPeers(3)
+				peers.Tongue = FakeDialer{}
+
+				go ConnectLoop(peers)
+				<-peers.maxedChan
+				So(peers.Count(), ShouldEqual, 3)
+
+				r := <-peers.snowflakeChan
+				So(peers.Count(), ShouldEqual, 2)
+				r.Close()
+				<-peers.maxedChan
+				So(peers.Count(), ShouldEqual, 3)
+
+				<-peers.snowflakeChan
+				<-peers.snowflakeChan
+				<-peers.snowflakeChan
+				So(peers.Count(), ShouldEqual, 0)
+			})
+		})
Convey("WebRTC Connection", func() {
    		c := new(webRTCConn)
@@ -60,17 +116,13 @@ func TestConnect(t *testing.T) {
    		}
    		So(c.buffer.Bytes(), ShouldEqual, nil)
-			Convey("Create and remove from WebRTCConn set", func() {
-				So(len(webrtcRemotes), ShouldEqual, 0)
-				So(remoteIndex, ShouldEqual, 0)
+			Convey("Can construct a WebRTCConn", func() {
    			s := NewWebRTCConnection(nil, nil)
    			So(s, ShouldNotBeNil)
    			So(s.index, ShouldEqual, 0)
-				So(len(webrtcRemotes), ShouldEqual, 1)
-				So(remoteIndex, ShouldEqual, 1)
+				So(s.offerChannel, ShouldNotBeNil)
+				So(s.answerChannel, ShouldNotBeNil)
    			s.Close()
-				So(len(webrtcRemotes), ShouldEqual, 0)
-				So(remoteIndex, ShouldEqual, 1)
    		})
Convey("Write buffers when datachannel is nil", func() {
@@ -113,9 +165,6 @@ func TestConnect(t *testing.T) {
    			<-c.reset
    		})
-			Convey("Connect Loop", func() {
-				// TODO
-			})
    	})
    })
@@ -124,14 +173,14 @@ func TestConnect(t *testing.T) {
    	transport := &MockTransport{http.StatusOK}
    	fakeOffer := webrtc.DeserializeSessionDescription("test")
-		Convey("BrokerChannel with no front domain", func() {
+		Convey("Construct BrokerChannel with no front domain", func() {
    		b := NewBrokerChannel("test.broker", "", transport)
    		So(b.url, ShouldNotBeNil)
    		So(b.url.Path, ShouldResemble, "test.broker")
    		So(b.transport, ShouldNotBeNil)
    	})
-		Convey("BrokerChannel with front domain", func() {
+		Convey("Construct BrokerChannel *with* front domain", func() {
    		b := NewBrokerChannel("test.broker", "front", transport)
    		So(b.url, ShouldNotBeNil)
    		So(b.url.Path, ShouldResemble, "test.broker")
@@ -139,7 +188,7 @@ func TestConnect(t *testing.T) {
    		So(b.transport, ShouldNotBeNil)
    	})
-		Convey("BrokerChannel Negotiate responds with answer", func() {
+		Convey("BrokerChannel.Negotiate responds with answer", func() {
    		b := NewBrokerChannel("test.broker", "", transport)
    		answer, err := b.Negotiate(fakeOffer)
    		So(err, ShouldBeNil)
@@ -147,7 +196,7 @@ func TestConnect(t *testing.T) {
    		So(answer.Sdp, ShouldResemble, "fake")
    	})
-		Convey("BrokerChannel Negotiate fails with 503", func() {
+		Convey("BrokerChannel.Negotiate fails with 503", func() {
    		b := NewBrokerChannel("test.broker", "",
    			&MockTransport{http.StatusServiceUnavailable})
    		answer, err := b.Negotiate(fakeOffer)
@@ -156,7 +205,7 @@ func TestConnect(t *testing.T) {
    		So(err.Error(), ShouldResemble, BrokerError503)
    	})
-		Convey("BrokerChannel Negotiate fails with 400", func() {
+		Convey("BrokerChannel.Negotiate fails with 400", func() {
    		b := NewBrokerChannel("test.broker", "",
    			&MockTransport{http.StatusBadRequest})
    		answer, err := b.Negotiate(fakeOffer)
@@ -165,7 +214,7 @@ func TestConnect(t *testing.T) {
    		So(err.Error(), ShouldResemble, BrokerError400)
    	})
-		Convey("BrokerChannel Negotiate fails with unexpected", func() {
+		Convey("BrokerChannel.Negotiate fails with unexpected error", func() {
    		b := NewBrokerChannel("test.broker", "",
    			&MockTransport{123})
    		answer, err := b.Negotiate(fakeOffer)
diff --git a/client/snowflake.go b/client/snowflake.go
index f32ddc8..61864ca 100644
--- a/client/snowflake.go
+++ b/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
 }
+// 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{}
+}
+
+func NewPeers(max int) *Peers {
+	p := &Peers{capacity: max}
+	p.snowflakeChan = make(chan *webRTCConn, max)
+	p.maxedChan = make(chan struct{}, 1)
+	return p
+}
+
+// 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")
+	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 SnowflakeConnectLoop() {
-	transport := CreateBrokerTransport()
-	broker = NewBrokerChannel(brokerURL, frontDomain, transport)
+func ConnectLoop(peers *Peers) {
    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()
+		s, err := peers.FindSnowflake()
    	if nil == s || nil != err {
-			log.Println("WebRTC Error: ", err, " retrying...")
+			log.Println("WebRTC Error:", err,
+				" Retrying in", ReconnectTimeout, "seconds...")
    		<-time.After(time.Second * ReconnectTimeout)
    		continue
    	}
-		snowflakeChan <- s
+		peers.snowflakeChan <- s
+		<-time.After(time.Second)
    }
 }
-// Initialize a WebRTC Connection.
-func dialWebRTC() (*webRTCConn, error) {
+// 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...)
-	if nil == broker {
-		return nil, errors.New("Failed to prepare BrokerChannel")
-	}
-	connection := NewWebRTCConnection(config, broker)
+	connection := NewWebRTCConnection(config, w.BrokerChannel)
    err := connection.Connect()
    return connection, err
 }
-func endWebRTC() {
-	log.Printf("WebRTC: interruped")
-	for _, r := range webrtcRemotes {
-		r.Close()
-	}
-}
-
 // 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 {
diff --git a/client/webrtc.go b/client/webrtc.go
index 5b30e95..e01cbf7 100644
--- a/client/webrtc.go
+++ b/client/webrtc.go
@@ -11,26 +11,27 @@ import (
    "time"
 )
-// Implements net.Conn interface
+// Remote WebRTC peer.  Implements the |net.Conn| interface.
 type webRTCConn struct {
-	config        *webrtc.Configuration
-	pc            *webrtc.PeerConnection
-	snowflake     SnowflakeChannel // Interface holding the WebRTC DataChannel.
-	broker        *BrokerChannel
+	config    *webrtc.Configuration
+	pc        *webrtc.PeerConnection
+	snowflake SnowflakeChannel // Holds the WebRTC DataChannel.
+	broker    *BrokerChannel
+
    offerChannel  chan *webrtc.SessionDescription
    answerChannel chan *webrtc.SessionDescription
    errorChannel  chan error
+	endChannel    chan struct{}
    recvPipe      *io.PipeReader
    writePipe     *io.PipeWriter
    buffer        bytes.Buffer
    reset         chan struct{}
-	index         int
+
+	index  int
+	closed bool
    *BytesInfo
 }
-var webrtcRemotes map[int]*webRTCConn
-var remoteIndex int = 0
-
 func (c *webRTCConn) Read(b []byte) (int, error) {
    return c.recvPipe.Read(b)
 }
@@ -51,10 +52,17 @@ func (c *webRTCConn) Close() error {
    var err error = nil
    log.Printf("WebRTC: Closing")
    c.cleanup()
-	close(c.offerChannel)
-	close(c.answerChannel)
-	close(c.errorChannel)
-	delete(webrtcRemotes, c.index)
+	if nil != c.offerChannel {
+		close(c.offerChannel)
+	}
+	if nil != c.answerChannel {
+		close(c.answerChannel)
+	}
+	if nil != c.errorChannel {
+		close(c.errorChannel)
+	}
+	// Mark for deletion.
+	c.closed = true
    return err
 }
@@ -78,6 +86,7 @@ func (c *webRTCConn) SetWriteDeadline(t time.Time) error {
    return fmt.Errorf("SetWriteDeadline not implemented")
 }
+// Construct a WebRTC PeerConnection.
 func NewWebRTCConnection(config *webrtc.Configuration,
    broker *BrokerChannel) *webRTCConn {
    connection := new(webRTCConn)
@@ -90,6 +99,7 @@ func NewWebRTCConnection(config *webrtc.Configuration,
    connection.errorChannel = make(chan error, 1)
    connection.reset = make(chan struct{}, 1)
+	// TODO: Separate out.
    // Log every few seconds.
    connection.BytesInfo = &BytesInfo{
    	inboundChan: make(chan int, 5), outboundChan: make(chan int, 5),
@@ -99,9 +109,6 @@ func NewWebRTCConnection(config *webrtc.Configuration,
// Pipes remain the same even when DataChannel gets switched.
    connection.recvPipe, connection.writePipe = io.Pipe()
-	connection.index = remoteIndex
-	webrtcRemotes[connection.index] = connection
-	remoteIndex++
    return connection
 }
@@ -296,12 +303,12 @@ func (c *webRTCConn) Reset() {
func (c *webRTCConn) cleanup() {
    if nil != c.snowflake {
-		s := c.snowflake
    	log.Printf("WebRTC: closing DataChannel")
+		dataChannel := c.snowflake
    	// Setting snowflake to nil *before* Close indicates to OnClose that it
    	// was locally triggered.
    	c.snowflake = nil
-		s.Close()
+		dataChannel.Close()
    }
    if nil != c.pc {
    	log.Printf("WebRTC: closing PeerConnection")

    

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

[tor-commits] [snowflake/master] client interfaces compose better, remove some globals, test ConnectLoop