notify/transport_binding.go

package notify

import (
	"b612.me/stario"
	"net"
	"sync"
	"time"
)

// transportBinding models the currently attached physical transport for a
// logical session. The binding can be swapped later without forcing callers to
// reach into raw conn fields directly.
type transportBinding struct {
	conn    net.Conn
	queue   *stario.StarQueue
	writeMu sync.Mutex

	controlMu     sync.Mutex
	controlSender *controlBatchSender

	bulkMu     sync.Mutex
	bulkSender *bulkBatchSender
}

func newTransportBinding(conn net.Conn, queue *stario.StarQueue) *transportBinding {
	if conn == nil && queue == nil {
		return nil
	}
	return &transportBinding{
		conn:  conn,
		queue: queue,
	}
}

func (b *transportBinding) connSnapshot() net.Conn {
	if b == nil {
		return nil
	}
	return b.conn
}

func (b *transportBinding) queueSnapshot() *stario.StarQueue {
	if b == nil {
		return nil
	}
	return b.queue
}

func (b *transportBinding) withConnWriteLock(fn func(net.Conn) error) error {
	return b.withConnWriteLockDeadline(time.Time{}, fn)
}

func (b *transportBinding) withConnWriteLockDeadline(deadline time.Time, fn func(net.Conn) error) error {
	if b == nil {
		return net.ErrClosed
	}
	b.writeMu.Lock()
	defer b.writeMu.Unlock()
	conn := b.connSnapshot()
	if conn == nil {
		return net.ErrClosed
	}
	if !deadline.IsZero() {
		if err := conn.SetWriteDeadline(deadline); err != nil {
			return err
		}
		defer func() {
			_ = conn.SetWriteDeadline(time.Time{})
		}()
	}
	return fn(conn)
}

func (b *transportBinding) bulkBatchSenderSnapshot() *bulkBatchSender {
	if b == nil {
		return nil
	}
	b.bulkMu.Lock()
	defer b.bulkMu.Unlock()
	if b.bulkSender != nil {
		return b.bulkSender
	}
	b.bulkSender = newBulkBatchSender(b)
	return b.bulkSender
}

func (b *transportBinding) controlBatchSenderSnapshot() *controlBatchSender {
	if b == nil {
		return nil
	}
	b.controlMu.Lock()
	defer b.controlMu.Unlock()
	if b.controlSender != nil {
		return b.controlSender
	}
	b.controlSender = newControlBatchSender(b)
	return b.controlSender
}

func (b *transportBinding) stopBackgroundWorkers() {
	if b == nil {
		return
	}
	b.controlMu.Lock()
	controlSender := b.controlSender
	b.controlMu.Unlock()
	b.bulkMu.Lock()
	bulkSender := b.bulkSender
	b.bulkMu.Unlock()
	if controlSender != nil {
		controlSender.stop()
	}
	if bulkSender != nil {
		bulkSender.stop()
	}
}