Style and minor fixes, main package

This commit is contained in:
Jakob Borg 2015-07-20 11:38:00 +02:00
parent 3d5507451b
commit ebef239a06
2 changed files with 52 additions and 24 deletions

View File

@ -27,7 +27,6 @@ func protocolListener(addr string, config *tls.Config) {
for { for {
conn, err := listener.Accept() conn, err := listener.Accept()
setTCPOptions(conn)
if err != nil { if err != nil {
if debug { if debug {
log.Println(err) log.Println(err)
@ -35,6 +34,8 @@ func protocolListener(addr string, config *tls.Config) {
continue continue
} }
setTCPOptions(conn)
if debug { if debug {
log.Println("Protocol listener accepted connection from", conn.RemoteAddr()) log.Println("Protocol listener accepted connection from", conn.RemoteAddr())
} }
@ -74,16 +75,12 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
errors := make(chan error, 1) errors := make(chan error, 1)
outbox := make(chan interface{}) outbox := make(chan interface{})
go func(conn net.Conn, message chan<- interface{}, errors chan<- error) { // Read messages from the connection and send them on the messages
for { // channel. When there is an error, send it on the error channel and
msg, err := protocol.ReadMessage(conn) // return. Applies also when the connection gets closed, so the pattern
if err != nil { // below is to close the connection on error, then wait for the error
errors <- err // signal from messageReader to exit.
return go messageReader(conn, messages, errors)
}
messages <- msg
}
}(conn, messages, errors)
pingTicker := time.NewTicker(pingInterval) pingTicker := time.NewTicker(pingInterval)
timeoutTicker := time.NewTimer(networkTimeout) timeoutTicker := time.NewTimer(networkTimeout)
@ -96,6 +93,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
if debug { if debug {
log.Printf("Message %T from %s", message, id) log.Printf("Message %T from %s", message, id)
} }
switch msg := message.(type) { switch msg := message.(type) {
case protocol.JoinRelayRequest: case protocol.JoinRelayRequest:
outboxesMut.RLock() outboxesMut.RLock()
@ -116,6 +114,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
joined = true joined = true
protocol.WriteMessage(conn, protocol.ResponseSuccess) protocol.WriteMessage(conn, protocol.ResponseSuccess)
case protocol.ConnectRequest: case protocol.ConnectRequest:
requestedPeer := syncthingprotocol.DeviceIDFromBytes(msg.ID) requestedPeer := syncthingprotocol.DeviceIDFromBytes(msg.ID)
outboxesMut.RLock() outboxesMut.RLock()
@ -151,7 +150,10 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
log.Println("Sent invitation from", id, "to", requestedPeer) log.Println("Sent invitation from", id, "to", requestedPeer)
} }
conn.Close() conn.Close()
case protocol.Pong: case protocol.Pong:
// Nothing
default: default:
if debug { if debug {
log.Printf("Unknown message %s: %T", id, message) log.Printf("Unknown message %s: %T", id, message)
@ -159,21 +161,25 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
protocol.WriteMessage(conn, protocol.ResponseUnexpectedMessage) protocol.WriteMessage(conn, protocol.ResponseUnexpectedMessage)
conn.Close() conn.Close()
} }
case err := <-errors: case err := <-errors:
if debug { if debug {
log.Printf("Closing connection %s: %s", id, err) log.Printf("Closing connection %s: %s", id, err)
} }
// Potentially closing a second time.
close(outbox) close(outbox)
// Potentially closing a second time.
conn.Close() conn.Close()
// Only delete the outbox if the client join, as it migth be a
// lookup request coming from the same client. // Only delete the outbox if the client is joined, as it might be
// a lookup request coming from the same client.
if joined { if joined {
outboxesMut.Lock() outboxesMut.Lock()
delete(outboxes, id) delete(outboxes, id)
outboxesMut.Unlock() outboxesMut.Unlock()
} }
return return
case <-pingTicker.C: case <-pingTicker.C:
if !joined { if !joined {
if debug { if debug {
@ -189,6 +195,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
} }
conn.Close() conn.Close()
} }
case <-timeoutTicker.C: case <-timeoutTicker.C:
// We should receive a error from the reader loop, which will cause // We should receive a error from the reader loop, which will cause
// us to quit this loop. // us to quit this loop.
@ -196,6 +203,7 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
log.Printf("%s timed out", id) log.Printf("%s timed out", id)
} }
conn.Close() conn.Close()
case msg := <-outbox: case msg := <-outbox:
if debug { if debug {
log.Printf("Sending message %T to %s", msg, id) log.Printf("Sending message %T to %s", msg, id)
@ -209,3 +217,14 @@ func protocolConnectionHandler(tcpConn net.Conn, config *tls.Config) {
} }
} }
} }
func messageReader(conn net.Conn, messages chan<- interface{}, errors chan<- error) {
for {
msg, err := protocol.ReadMessage(conn)
if err != nil {
errors <- err
return
}
messages <- msg
}
}

View File

@ -18,7 +18,6 @@ func sessionListener(addr string) {
for { for {
conn, err := listener.Accept() conn, err := listener.Accept()
setTCPOptions(conn)
if err != nil { if err != nil {
if debug { if debug {
log.Println(err) log.Println(err)
@ -26,6 +25,8 @@ func sessionListener(addr string) {
continue continue
} }
setTCPOptions(conn)
if debug { if debug {
log.Println("Session listener accepted connection from", conn.RemoteAddr()) log.Println("Session listener accepted connection from", conn.RemoteAddr())
} }
@ -35,10 +36,17 @@ func sessionListener(addr string) {
} }
func sessionConnectionHandler(conn net.Conn) { func sessionConnectionHandler(conn net.Conn) {
conn.SetDeadline(time.Now().Add(messageTimeout)) defer conn.Close()
if err := conn.SetDeadline(time.Now().Add(messageTimeout)); err != nil {
if debug {
log.Println("Weird error setting deadline:", err, "on", conn.RemoteAddr())
}
return
}
message, err := protocol.ReadMessage(conn) message, err := protocol.ReadMessage(conn)
if err != nil { if err != nil {
conn.Close()
return return
} }
@ -51,7 +59,6 @@ func sessionConnectionHandler(conn net.Conn) {
if ses == nil { if ses == nil {
protocol.WriteMessage(conn, protocol.ResponseNotFound) protocol.WriteMessage(conn, protocol.ResponseNotFound)
conn.Close()
return return
} }
@ -60,24 +67,26 @@ func sessionConnectionHandler(conn net.Conn) {
log.Println("Failed to add", conn.RemoteAddr(), "to session", ses) log.Println("Failed to add", conn.RemoteAddr(), "to session", ses)
} }
protocol.WriteMessage(conn, protocol.ResponseAlreadyConnected) protocol.WriteMessage(conn, protocol.ResponseAlreadyConnected)
conn.Close()
return return
} }
err := protocol.WriteMessage(conn, protocol.ResponseSuccess) if err := protocol.WriteMessage(conn, protocol.ResponseSuccess); err != nil {
if err != nil {
if debug { if debug {
log.Println("Failed to send session join response to ", conn.RemoteAddr(), "for", ses) log.Println("Failed to send session join response to ", conn.RemoteAddr(), "for", ses)
} }
conn.Close()
return return
} }
conn.SetDeadline(time.Time{})
if err := conn.SetDeadline(time.Time{}); err != nil {
if debug {
log.Println("Weird error setting deadline:", err, "on", conn.RemoteAddr())
}
return
}
default: default:
if debug { if debug {
log.Println("Unexpected message from", conn.RemoteAddr(), message) log.Println("Unexpected message from", conn.RemoteAddr(), message)
} }
protocol.WriteMessage(conn, protocol.ResponseUnexpectedMessage) protocol.WriteMessage(conn, protocol.ResponseUnexpectedMessage)
conn.Close()
} }
} }