<commit_before>package transport

import (
	"time"

	"github.com/stefankopieczek/gossip/log"
	"github.com/stefankopieczek/gossip/timing"
)

// Fields of connTable should only be modified by the dedicated goroutine called by Init().
// All other callers should use connTable's associated public methods to access it.
type connTable struct {
	conns        map[string]*connWatcher
	connRequests chan *connRequest
	updates      chan *connUpdate
	expiries     chan string
	stop         chan bool
	stopped      bool
}

type connWatcher struct {
	addr       string
	conn       *connection
	timer      timing.Timer
	expiryTime time.Time
	expiry     chan<- string
	stop       chan bool
}

// Create a new connection table.
func (t *connTable) Init() {
	log.Info("Init conntable %p", t)
	t.conns = make(map[string]*connWatcher)
	t.connRequests = make(chan *connRequest)
	t.updates = make(chan *connUpdate)
	t.expiries = make(chan string)
	t.stop = make(chan bool)
	go t.manage()
}

// Management loop for the connTable.
// Handles notifications of connection updates, expiries of connections, and
// the termination of the routine.
func (t *connTable) manage() {
	for {
		select {
		case request := <-t.connRequests:
			watcher := t.conns[request.addr]
			if watcher != nil {
				request.responseChan <- watcher.conn
			} else {
				request.responseChan <- nil
			}
		case update := <-t.updates:
			t.handleUpdate(update)
		case addr := <-t.expiries:
			if t.conns[addr].expiryTime.Before(time.Now()) {
				log.Debug("Conntable %p notified that the watcher for address %s has expired. Remove it.", t, addr)
				t.conns[addr].stop <- true
				t.conns[addr].conn.Close()
				delete(t.conns, addr)
			} else {
                // Due to a race condition, the socket has been updated since this expiry happened.
                // Ignore the expiry since we already have a new socket for this address.
            }
		case <-t.stop:
			log.Info("Conntable %p stopped")
			t.stopped = true
			for _, watcher := range t.conns {
				watcher.stop <- true
				watcher.conn.Close()
			}
			break
		}
	}
}

// Push a connection to the connection table, registered under a specific address.
// If it is a new connection, start the socket expiry timer.<commit_msg>Issue 13 - Panic in conntable - add logging<commit_after>package transport

import (
	"time"

	"github.com/stefankopieczek/gossip/log"
	"github.com/stefankopieczek/gossip/timing"
)

// Fields of connTable should only be modified by the dedicated goroutine called by Init().
// All other callers should use connTable's associated public methods to access it.
type connTable struct {
	conns        map[string]*connWatcher
	connRequests chan *connRequest
	updates      chan *connUpdate
	expiries     chan string
	stop         chan bool
	stopped      bool
}

type connWatcher struct {
	addr       string
	conn       *connection
	timer      timing.Timer
	expiryTime time.Time
	expiry     chan<- string
	stop       chan bool
}

// Create a new connection table.
func (t *connTable) Init() {
	log.Info("Init conntable %p", t)
	t.conns = make(map[string]*connWatcher)
	t.connRequests = make(chan *connRequest)
	t.updates = make(chan *connUpdate)
	t.expiries = make(chan string)
	t.stop = make(chan bool)
	go t.manage()
}

// Management loop for the connTable.
// Handles notifications of connection updates, expiries of connections, and
// the termination of the routine.
func (t *connTable) manage() {
	for {
		select {
		case request := <-t.connRequests:
			watcher := t.conns[request.addr]
			if watcher != nil {
				request.responseChan <- watcher.conn
			} else {
				request.responseChan <- nil
			}
		case update := <-t.updates:
			t.handleUpdate(update)
		case addr := <-t.expiries:
			if t.conns[addr].expiryTime.Before(time.Now()) {
				log.Debug("Conntable %p notified that the watcher for address %s has expired. Remove it.", t, addr)
				t.conns[addr].stop <- true
				t.conns[addr].conn.Close()
				delete(t.conns, addr)
			} else {
                // Due to a race condition, the socket has been updated since this expiry happened.
                // Ignore the expiry since we already have a new socket for this address.
                log.Warn("Ignored spurious expiry for address %s in conntable %p", t, addr)
            }
		case <-t.stop:
			log.Info("Conntable %p stopped")
			t.stopped = true
			for _, watcher := range t.conns {
				watcher.stop <- true
				watcher.conn.Close()
			}
			break
		}
	}
}

// Push a connection to the connection table, registered under a specific address.
// If it is a new connection, start the socket expiry timer.
{'fraction_non_alphanumeric': 0.10234680573663625, 'fraction_numerical': 0.000434593654932638, 'mean_word_length': 2.8713204373423045, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2846333', 'n_tokens_mistral': 1631, 'n_tokens_neox': 1441, 'n_words': 476}