closer (#132)

closer.go

@@ -1,6 +1,43 @@
 package dilithium
 
-// Closer manages the state machine for the shutdown of a TxPortal and RxPortal pair (one side of a communication).
+import (
+	"github.com/openziti/dilithium/util"
+	"github.com/sirupsen/logrus"
+	"time"
+)
+
+// closer manages the state machine for the shutdown of a TxPortal and RxPortal pair (one side of a communication).
 //
-type Closer struct {
+type closer struct {
+	seq          *util.Sequence
+	rxCloseSeq   int32
+	rxCloseSeqIn chan int32
+	txCloseSeq   int32
+	txCloseSeqIn chan int32
+	txp          *TxPortal
+	rxp          *RxPortal
+	lastEvent    time.Time
+	closeHook    func()
+}
+
+func newCloser(seq *util.Sequence, closeHook func()) *closer {
+	return &closer{
+		seq:          seq,
+		rxCloseSeq:   notClosed,
+		rxCloseSeqIn: make(chan int32, 1),
+		txCloseSeq:   notClosed,
+		txCloseSeqIn: make(chan int32, 1),
+		closeHook:    closeHook,
+	}
 }
+
+func (c *closer) emergencyStop() {
+	logrus.Info("broken glass")
+	c.txp.close()
+	c.rxp.Close()
+	if c.closeHook != nil {
+		c.closeHook()
+	}
+}
+
+const notClosed = int32(-99)

rxportal.go

@@ -25,7 +25,7 @@ type RxPortal struct {
 	ackPool      *Pool
 	txp          *TxPortal
 	seq          *util.Sequence
-	closer       *Closer
+	closer       *closer
 	closed       bool
 }
 
@@ -35,7 +35,7 @@ type RxRead struct {
 	Eof  bool
 }
 
-func NewRxPortal(transport Transport, txp *TxPortal, seq *util.Sequence, closer *Closer) *RxPortal {
+func NewRxPortal(transport Transport, txp *TxPortal, seq *util.Sequence, closer *closer) *RxPortal {
 	rxp := &RxPortal{
 		transport:  transport,
 		tree:       btree.NewWith(txp.alg.Profile().MaxTreeSize, utils.Int32Comparator),

txportal.go

@@ -23,12 +23,12 @@ type TxPortal struct {
 	transport Transport
 	alg       TxAlgorithm
 	monitor   *TxMonitor
-	closer    *Closer
+	closer    *closer
 	closed    bool
 	pool      *Pool
 }
 
-func newTxPortal(transport Transport, alg TxAlgorithm, closer *Closer, pool *Pool) *TxPortal {
+func newTxPortal(transport Transport, alg TxAlgorithm, closer *closer, pool *Pool) *TxPortal {
 	txp := &TxPortal{
 		lock:      new(sync.Mutex),
 		tree:      btree.NewWith(alg.Profile().MaxTreeSize, utils.Int32Comparator),
@@ -127,6 +127,11 @@ func (txp *TxPortal) ack(acks []Ack) error {
 	return nil
 }
 
+func (txp *TxPortal) close() {
+	txp.closed = true
+	txp.monitor.close()
+}
+
 func (txp *TxPortal) keepaliveSender() {
 	logrus.Info("started")
 	defer logrus.Info("exited")