Merge pull request #116 from 9seconds/gcoldconnections

Forcefully cleanup old connections

hub/connection_list.go

@@ -11,11 +11,7 @@ type connectionList struct {
 	connections []*connection
 }
 
-func (c *connectionList) Get(conn *ProxyConn) (*connection, error) {
-	if len(c.connections) > 0 {
-		c.gc()
-	}
-
+func (c *connectionList) get(conn *ProxyConn) (*connection, error) {
 	if len(c.connections) > 0 && c.connections[0].Len() < config.C.MultiplexPerConnection {
 		if err := c.connections[0].Attach(conn); err == nil {
 			return c.connections[0], nil
@@ -41,6 +37,9 @@ func (c *connectionList) Get(conn *ProxyConn) (*connection, error) {
 
 func (c *connectionList) gc() {
 	prevLen := len(c.connections)
+	if prevLen == 0 {
+		return
+	}
 
 	for i := len(c.connections) - 1; i >= 0; i-- {
 		lastIndex := len(c.connections) - 1

hub/mux.go

@@ -2,11 +2,14 @@ package hub
 
 import (
 	"context"
+	"time"
 
 	"mtg/conntypes"
 	"mtg/protocol"
 )
 
+const muxGCEvery = time.Minute
+
 type muxNewRequest struct {
 	req  *protocol.TelegramRequest
 	resp chan<- muxNewResponse
@@ -26,6 +29,9 @@ type mux struct {
 }
 
 func (m *mux) run() {
+	gcTicker := time.NewTicker(muxGCEvery)
+	defer gcTicker.Stop()
+
 	for {
 		select {
 		case <-m.ctx.Done():
@@ -34,9 +40,12 @@ func (m *mux) run() {
 			}
 
 			return
+		case <-gcTicker.C:
+			m.connections.gc()
 		case req := <-m.channelNew:
+			m.connections.gc()
 			proxyConn := newProxyConn(req.req, m.channelClosed)
-			conn, err := m.connections.Get(proxyConn)
+			conn, err := m.connections.get(proxyConn)
 
 			if err == nil {
 				m.clients[string(req.req.ConnID[:])] = conn