Index: trunk/conn.go
===================================================================
--- trunk/conn.go	(revision 210)
+++ trunk/conn.go	(revision 210)
@@ -0,0 +1,109 @@
+package soju
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	"gopkg.in/irc.v3"
+)
+
+func setKeepAlive(c net.Conn) error {
+	tcpConn, ok := c.(*net.TCPConn)
+	if !ok {
+		return fmt.Errorf("cannot enable keep-alive on a non-TCP connection")
+	}
+	if err := tcpConn.SetKeepAlive(true); err != nil {
+		return err
+	}
+	return tcpConn.SetKeepAlivePeriod(keepAlivePeriod)
+}
+
+type conn struct {
+	net      net.Conn
+	irc      *irc.Conn
+	srv      *Server
+	logger   Logger
+	outgoing chan<- *irc.Message
+	closed   chan struct{}
+}
+
+func newConn(srv *Server, netConn net.Conn, logger Logger) *conn {
+	setKeepAlive(netConn)
+
+	outgoing := make(chan *irc.Message, 64)
+	c := &conn{
+		net:      netConn,
+		irc:      irc.NewConn(netConn),
+		srv:      srv,
+		outgoing: outgoing,
+		logger:   logger,
+		closed:   make(chan struct{}),
+	}
+
+	go func() {
+		for msg := range outgoing {
+			if c.srv.Debug {
+				c.logger.Printf("sent: %v", msg)
+			}
+			c.net.SetWriteDeadline(time.Now().Add(writeTimeout))
+			if err := c.irc.WriteMessage(msg); err != nil {
+				c.logger.Printf("failed to write message: %v", err)
+				break
+			}
+		}
+		if err := c.net.Close(); err != nil {
+			c.logger.Printf("failed to close connection: %v", err)
+		} else {
+			c.logger.Printf("connection closed")
+		}
+		// Drain the outgoing channel to prevent SendMessage from blocking
+		for range outgoing {
+			// This space is intentionally left blank
+		}
+	}()
+
+	c.logger.Printf("new connection")
+	return c
+}
+
+func (c *conn) isClosed() bool {
+	select {
+	case <-c.closed:
+		return true
+	default:
+		return false
+	}
+}
+
+// Close closes the connection. It is safe to call from any goroutine.
+func (c *conn) Close() error {
+	if c.isClosed() {
+		return fmt.Errorf("connection already closed")
+	}
+	close(c.closed)
+	close(c.outgoing)
+	return nil
+}
+
+func (c *conn) ReadMessage() (*irc.Message, error) {
+	msg, err := c.irc.ReadMessage()
+	if err != nil {
+		return nil, err
+	}
+
+	if c.srv.Debug {
+		c.logger.Printf("received: %v", msg)
+	}
+
+	return msg, nil
+}
+
+// SendMessage queues a new outgoing message. It is safe to call from any
+// goroutine.
+func (c *conn) SendMessage(msg *irc.Message) {
+	if c.isClosed() {
+		return
+	}
+	c.outgoing <- msg
+}
Index: trunk/downstream.go
===================================================================
--- trunk/downstream.go	(revision 209)
+++ trunk/downstream.go	(revision 210)
@@ -53,11 +53,7 @@
 
 type downstreamConn struct {
-	id       uint64
-	net      net.Conn
-	irc      *irc.Conn
-	srv      *Server
-	logger   Logger
-	outgoing chan<- *irc.Message
-	closed   chan struct{}
+	conn
+
+	id uint64
 
 	registered  bool
@@ -85,13 +81,8 @@
 
 func newDownstreamConn(srv *Server, netConn net.Conn, id uint64) *downstreamConn {
-	outgoing := make(chan *irc.Message, 64)
+	logger := &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())}
 	dc := &downstreamConn{
+		conn:          *newConn(srv, netConn, logger),
 		id:            id,
-		net:           netConn,
-		irc:           irc.NewConn(netConn),
-		srv:           srv,
-		logger:        &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
-		outgoing:      outgoing,
-		closed:        make(chan struct{}),
 		ringConsumers: make(map[*network]*RingConsumer),
 		caps:          make(map[string]bool),
@@ -102,28 +93,4 @@
 		dc.hostname = host
 	}
-
-	go func() {
-		for msg := range outgoing {
-			if dc.srv.Debug {
-				dc.logger.Printf("sent: %v", msg)
-			}
-			dc.net.SetWriteDeadline(time.Now().Add(writeTimeout))
-			if err := dc.irc.WriteMessage(msg); err != nil {
-				dc.logger.Printf("failed to write message: %v", err)
-				break
-			}
-		}
-		if err := dc.net.Close(); err != nil {
-			dc.logger.Printf("failed to close connection: %v", err)
-		} else {
-			dc.logger.Printf("connection closed")
-		}
-		// Drain the outgoing channel to prevent SendMessage from blocking
-		for range outgoing {
-			// This space is intentionally left blank
-		}
-	}()
-
-	dc.logger.Printf("new connection")
 	return dc
 }
@@ -228,16 +195,7 @@
 }
 
-func (dc *downstreamConn) isClosed() bool {
-	select {
-	case <-dc.closed:
-		return true
-	default:
-		return false
-	}
-}
-
 func (dc *downstreamConn) readMessages(ch chan<- event) error {
 	for {
-		msg, err := dc.irc.ReadMessage()
+		msg, err := dc.ReadMessage()
 		if err == io.EOF {
 			break
@@ -246,8 +204,4 @@
 		}
 
-		if dc.srv.Debug {
-			dc.logger.Printf("received: %v", msg)
-		}
-
 		ch <- eventDownstreamMessage{msg, dc}
 	}
@@ -256,26 +210,7 @@
 }
 
-func (dc *downstreamConn) writeMessages() error {
-	return nil
-}
-
-// Close closes the connection. It is safe to call from any goroutine.
-func (dc *downstreamConn) Close() error {
-	if dc.isClosed() {
-		return fmt.Errorf("downstream connection already closed")
-	}
-	close(dc.closed)
-	close(dc.outgoing)
-	return nil
-}
-
-// SendMessage queues a new outgoing message. It is safe to call from any
-// goroutine.
 func (dc *downstreamConn) SendMessage(msg *irc.Message) {
-	if dc.isClosed() {
-		return
-	}
 	// TODO: strip tags if the client doesn't support them (see runNetwork)
-	dc.outgoing <- msg
+	dc.conn.SendMessage(msg)
 }
 
Index: trunk/server.go
===================================================================
--- trunk/server.go	(revision 209)
+++ trunk/server.go	(revision 210)
@@ -16,15 +16,4 @@
 var connectTimeout = 15 * time.Second
 var writeTimeout = 10 * time.Second
-
-func setKeepAlive(c net.Conn) error {
-	tcpConn, ok := c.(*net.TCPConn)
-	if !ok {
-		return fmt.Errorf("cannot enable keep-alive on a non-TCP connection")
-	}
-	if err := tcpConn.SetKeepAlive(true); err != nil {
-		return err
-	}
-	return tcpConn.SetKeepAlivePeriod(keepAlivePeriod)
-}
 
 type Logger interface {
@@ -110,6 +99,4 @@
 		}
 
-		setKeepAlive(netConn)
-
 		dc := newDownstreamConn(s, netConn, nextDownstreamID)
 		nextDownstreamID++
Index: trunk/upstream.go
===================================================================
--- trunk/upstream.go	(revision 209)
+++ trunk/upstream.go	(revision 210)
@@ -32,12 +32,8 @@
 
 type upstreamConn struct {
-	network  *network
-	logger   Logger
-	net      net.Conn
-	irc      *irc.Conn
-	srv      *Server
-	user     *user
-	outgoing chan<- *irc.Message
-	closed   chan struct{}
+	conn
+
+	network *network
+	user    *user
 
 	serverName            string
@@ -91,16 +87,8 @@
 	}
 
-	setKeepAlive(netConn)
-
-	outgoing := make(chan *irc.Message, 64)
 	uc := &upstreamConn{
+		conn:                     *newConn(network.user.srv, netConn, logger),
 		network:                  network,
-		logger:                   logger,
-		net:                      netConn,
-		irc:                      irc.NewConn(netConn),
-		srv:                      network.user.srv,
 		user:                     network.user,
-		outgoing:                 outgoing,
-		closed:                   make(chan struct{}),
 		channels:                 make(map[string]*upstreamChannel),
 		caps:                     make(map[string]string),
@@ -113,46 +101,5 @@
 	}
 
-	go func() {
-		for msg := range outgoing {
-			if uc.srv.Debug {
-				uc.logger.Printf("sent: %v", msg)
-			}
-			uc.net.SetWriteDeadline(time.Now().Add(writeTimeout))
-			if err := uc.irc.WriteMessage(msg); err != nil {
-				uc.logger.Printf("failed to write message: %v", err)
-				break
-			}
-		}
-		if err := uc.net.Close(); err != nil {
-			uc.logger.Printf("failed to close connection: %v", err)
-		} else {
-			uc.logger.Printf("connection closed")
-		}
-		// Drain the outgoing channel to prevent SendMessage from blocking
-		for range outgoing {
-			// This space is intentionally left blank
-		}
-	}()
-
 	return uc, nil
-}
-
-func (uc *upstreamConn) isClosed() bool {
-	select {
-	case <-uc.closed:
-		return true
-	default:
-		return false
-	}
-}
-
-// Close closes the connection. It is safe to call from any goroutine.
-func (uc *upstreamConn) Close() error {
-	if uc.isClosed() {
-		return fmt.Errorf("upstream connection already closed")
-	}
-	close(uc.closed)
-	close(uc.outgoing)
-	return nil
 }
 
@@ -1410,5 +1357,5 @@
 func (uc *upstreamConn) readMessages(ch chan<- event) error {
 	for {
-		msg, err := uc.irc.ReadMessage()
+		msg, err := uc.ReadMessage()
 		if err == io.EOF {
 			break
@@ -1417,18 +1364,8 @@
 		}
 
-		if uc.srv.Debug {
-			uc.logger.Printf("received: %v", msg)
-		}
-
 		ch <- eventUpstreamMessage{msg, uc}
 	}
 
 	return nil
-}
-
-// SendMessage queues a new outgoing message. It is safe to call from any
-// goroutine.
-func (uc *upstreamConn) SendMessage(msg *irc.Message) {
-	uc.outgoing <- msg
 }
 
