Commit a71c98c0 authored by Serene Han's avatar Serene Han
Browse files

able to break out of ConnectLoop, try separate webrtcConfigs as well

parent 4ca0a3aa
......@@ -65,8 +65,9 @@ func (f FakeSocksConn) Grant(addr *net.TCPAddr) error { return nil }
type FakePeers struct{ toRelease *webRTCConn }
func (f FakePeers) Collect() error { return nil }
func (f FakePeers) Pop() Snowflake { return nil }
func (f FakePeers) Collect() error { return nil }
func (f FakePeers) Pop() Snowflake { return nil }
func (f FakePeers) Melted() <-chan struct{} { return nil }
func TestSnowflakeClient(t *testing.T) {
......@@ -144,6 +145,7 @@ func TestSnowflakeClient(t *testing.T) {
}
So(p.Count(), ShouldEqual, cnt)
p.End()
<-p.Melted()
So(p.Count(), ShouldEqual, 0)
})
......@@ -168,7 +170,6 @@ func TestSnowflakeClient(t *testing.T) {
Convey("Can construct a WebRTCConn", func() {
s := NewWebRTCConnection(nil, nil)
So(s, ShouldNotBeNil)
So(s.index, ShouldEqual, 0)
So(s.offerChannel, ShouldNotBeNil)
So(s.answerChannel, ShouldNotBeNil)
s.Close()
......@@ -176,13 +177,13 @@ func TestSnowflakeClient(t *testing.T) {
Convey("Write buffers when datachannel is nil", func() {
c.Write([]byte("test"))
c.snowflake = nil
c.transport = nil
So(c.buffer.Bytes(), ShouldResemble, []byte("test"))
})
Convey("Write sends to datachannel when not nil", func() {
mock := new(MockDataChannel)
c.snowflake = mock
c.transport = mock
mock.done = make(chan bool, 1)
c.Write([]byte("test"))
<-mock.done
......
......@@ -38,6 +38,9 @@ type SnowflakeCollector interface {
// Remove and return the most available Snowflake from the collection.
Pop() Snowflake
// Signal when the collector has stopped collecting.
Melted() <-chan struct{}
}
// Interface to adapt to goptlib's SocksConn struct.
......
......@@ -25,6 +25,8 @@ type Peers struct {
snowflakeChan chan Snowflake
activePeers *list.List
capacity int
melt chan struct{}
}
// Construct a fresh container of remote peers.
......@@ -33,17 +35,19 @@ func NewPeers(max int) *Peers {
// Use buffered go channel to pass snowflakes onwards to the SOCKS handler.
p.snowflakeChan = make(chan Snowflake, max)
p.activePeers = list.New()
p.melt = make(chan struct{}, 1)
return p
}
// As part of |SnowflakeCollector| interface.
func (p *Peers) Collect() error {
cnt := p.Count()
s := fmt.Sprintf("Currently at [%d/%d]", cnt, p.capacity)
if cnt >= p.capacity {
s := fmt.Sprintf("At capacity [%d/%d]", cnt, p.capacity)
return errors.New(s)
}
log.Println("WebRTC: Collecting a new Snowflake.", s)
// Engage the Snowflake Catching interface, which must be available.
if nil == p.Tongue {
return errors.New("Missing Tongue to catch Snowflakes with.")
......@@ -60,7 +64,6 @@ func (p *Peers) Collect() error {
// As part of |SnowflakeCollector| interface.
func (p *Peers) Pop() Snowflake {
// Blocks until an available snowflake appears.
snowflake, ok := <-p.snowflakeChan
if !ok {
......@@ -71,6 +74,11 @@ func (p *Peers) Pop() Snowflake {
return snowflake
}
// As part of |SnowflakeCollector| interface.
func (p *Peers) Melted() <-chan struct{} {
return p.melt
}
// Returns total available Snowflakes (including the active one)
// The count only reduces when connections themselves close, rather than when
// they are popped.
......@@ -93,9 +101,16 @@ func (p *Peers) purgeClosedPeers() {
// Close all Peers contained here.
func (p *Peers) End() {
log.Printf("WebRTC: Ending all peer connections.")
for e := p.activePeers.Front(); e != nil; e = e.Next() {
close(p.snowflakeChan)
p.melt <- struct{}{}
cnt := p.Count()
for e := p.activePeers.Front(); e != nil; {
log.Println(e, e.Value)
next := e.Next()
conn := e.Value.(*webRTCConn)
conn.Close()
p.activePeers.Remove(e)
e = next
}
log.Println("WebRTC: melted all", cnt, "snowflakes.")
}
......@@ -115,15 +115,17 @@ func (bc *BrokerChannel) Negotiate(offer *webrtc.SessionDescription) (
// Implements the |Tongue| interface to catch snowflakes, using BrokerChannel.
type WebRTCDialer struct {
*BrokerChannel
webrtcConfig *webrtc.Configuration
// webrtcConfig *webrtc.Configuration
iceServers IceServerList
}
func NewWebRTCDialer(
broker *BrokerChannel, iceServers IceServerList) *WebRTCDialer {
config := webrtc.NewConfiguration(iceServers...)
return &WebRTCDialer{
BrokerChannel: broker,
webrtcConfig: config,
iceServers: iceServers,
// webrtcConfig: config,
}
}
......@@ -134,7 +136,8 @@ func (w WebRTCDialer) Catch() (Snowflake, error) {
}
// TODO: [#3] Fetch ICE server information from Broker.
// TODO: [#18] Consider TURN servers here too.
connection := NewWebRTCConnection(w.webrtcConfig, w.BrokerChannel)
config := webrtc.NewConfiguration(w.iceServers...)
connection := NewWebRTCConnection(config, w.BrokerChannel)
err := connection.Connect()
return connection, err
}
......
......@@ -30,17 +30,19 @@ var handlerChan = make(chan int)
// transfer to the Tor SOCKS handler when needed.
func ConnectLoop(snowflakes SnowflakeCollector) {
for {
// Check if ending is necessary.
err := snowflakes.Collect()
if nil != err {
log.Println("WebRTC:", err,
" Retrying in", ReconnectTimeout, "seconds...")
// Failed collections get a timeout.
<-time.After(time.Second * ReconnectTimeout)
}
select {
case <-time.After(time.Second * ReconnectTimeout):
continue
case <-snowflakes.Melted():
log.Println("ConnectLoop: stopped.")
return
}
// Successful collection gets rate limited to once per second.
log.Println("WebRTC: Connected to new Snowflake.")
<-time.After(time.Second)
}
}
......@@ -50,7 +52,7 @@ func socksAcceptLoop(ln *pt.SocksListener, snowflakes SnowflakeCollector) error
log.Println("Started SOCKS listener.")
for {
conn, err := ln.AcceptSocks()
log.Println("SOCKS accepted ", conn.Req)
log.Println("SOCKS accepted: ", conn.Req)
if err != nil {
if e, ok := err.(net.Error); ok && e.Temporary() {
continue
......@@ -72,6 +74,7 @@ func handler(socks SocksConnector, snowflakes SnowflakeCollector) error {
handlerChan <- -1
}()
// Obtain an available WebRTC remote. May block.
log.Println("handler: awaiting Snowflake...")
snowflake := snowflakes.Pop()
if nil == snowflake {
socks.Reject()
......@@ -85,6 +88,7 @@ func handler(socks SocksConnector, snowflakes SnowflakeCollector) error {
}
// Begin exchanging data.
// BUG(serene): There's a leak here when multiplexed.
go copyLoop(socks, snowflake)
// When WebRTC resets, close the SOCKS connection, which induces new handler.
......
......@@ -15,19 +15,17 @@ import (
type webRTCConn struct {
config *webrtc.Configuration
pc *webrtc.PeerConnection
snowflake SnowflakeDataChannel // Holds the WebRTC DataChannel.
transport SnowflakeDataChannel // 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
closed bool
BytesLogger
......@@ -43,11 +41,11 @@ func (c *webRTCConn) Read(b []byte) (int, error) {
// As part of |io.ReadWriter|
func (c *webRTCConn) Write(b []byte) (int, error) {
c.BytesLogger.AddOutbound(len(b))
if nil == c.snowflake {
if nil == c.transport {
log.Printf("Buffered %d bytes --> WebRTC", len(b))
c.buffer.Write(b)
} else {
c.snowflake.Send(b)
c.transport.Send(b)
}
return len(b), nil
}
......@@ -57,15 +55,6 @@ func (c *webRTCConn) Close() error {
var err error = nil
log.Printf("WebRTC: Closing")
c.cleanup()
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
......@@ -106,7 +95,6 @@ func NewWebRTCConnection(config *webrtc.Configuration,
// As part of |Connector| interface.
func (c *webRTCConn) Connect() error {
log.Printf("Establishing WebRTC connection #%d...", c.index)
// TODO: When go-webrtc is more stable, it's possible that a new
// PeerConnection won't need to be re-prepared each time.
err := c.preparePeerConnection()
......@@ -174,7 +162,7 @@ func (c *webRTCConn) preparePeerConnection() error {
// Create a WebRTC DataChannel locally.
func (c *webRTCConn) establishDataChannel() error {
if c.snowflake != nil {
if c.transport != nil {
panic("Unexpected datachannel already exists!")
}
dc, err := c.pc.CreateDataChannel("snowflake", webrtc.Init{})
......@@ -187,9 +175,8 @@ func (c *webRTCConn) establishDataChannel() error {
}
dc.OnOpen = func() {
log.Println("WebRTC: DataChannel.OnOpen")
if nil != c.snowflake {
log.Println("PeerConnection snowflake already exists.")
panic("PeerConnection snowflake already exists.")
if nil != c.transport {
panic("WebRTC: transport already exists.")
}
// Flush buffered outgoing SOCKS data if necessary.
if c.buffer.Len() > 0 {
......@@ -198,11 +185,11 @@ func (c *webRTCConn) establishDataChannel() error {
c.buffer.Reset()
}
// Then enable the datachannel.
c.snowflake = dc
c.transport = dc
}
dc.OnClose = func() {
// Future writes will go to the buffer until a new DataChannel is available.
if nil == c.snowflake {
if nil == c.transport {
// Closed locally, as part of a reset.
log.Println("WebRTC: DataChannel.OnClose [locally]")
return
......@@ -210,7 +197,7 @@ func (c *webRTCConn) establishDataChannel() error {
// Closed remotely, need to reset everything.
// Disable the DataChannel as a write destination.
log.Println("WebRTC: DataChannel.OnClose [remotely]")
c.snowflake = nil
c.transport = nil
c.Reset()
}
dc.OnMessage = func(msg []byte) {
......@@ -284,13 +271,23 @@ func (c *webRTCConn) exchangeSDP() error {
return nil
}
// Close all channels and transports
func (c *webRTCConn) cleanup() {
if nil != c.snowflake {
if nil != c.offerChannel {
close(c.offerChannel)
}
if nil != c.answerChannel {
close(c.answerChannel)
}
if nil != c.errorChannel {
close(c.errorChannel)
}
if nil != c.transport {
log.Printf("WebRTC: closing DataChannel")
dataChannel := c.snowflake
// Setting snowflake to nil *before* Close indicates to OnClose that it
dataChannel := c.transport
// Setting dc to nil *before* Close indicates to OnClose that it
// was locally triggered.
c.snowflake = nil
c.transport = nil
dataChannel.Close()
}
if nil != c.pc {
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment