johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package relay

import (
	"sync"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/clock"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt"
)

const (
	DefaultMaxBacklog      = 10000
	DefaultMaxBacklogBytes = 64 * 1024 * 1024
)

type Sequencer struct {
	mu       sync.Mutex
	seq      int64
	max      int
	maxBytes int
	bytes    int
	backlog  []Frame
	sizes    []int
	subs     map[uint64]chan Frame
	nextSub  uint64
}

func NewSequencer(maxBacklog int) *Sequencer {
	maxBacklog = opt.Positive(maxBacklog, DefaultMaxBacklog)
	return &Sequencer{
		max:      maxBacklog,
		maxBytes: DefaultMaxBacklogBytes,
		subs:     map[uint64]chan Frame{},
	}
}

func (s *Sequencer) Append(frame Frame) Frame {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.seq++
	frame.Seq = s.seq
	frame.Time = clock.NowISO()
	size := len(frame.Blocks) + 512
	s.backlog = append(s.backlog, frame)
	s.sizes = append(s.sizes, size)
	s.bytes += size
	for len(s.backlog) > s.max || (s.bytes > s.maxBytes && len(s.backlog) > 1) {
		s.bytes -= s.sizes[0]
		s.backlog = s.backlog[1:]
		s.sizes = s.sizes[1:]
	}
	for id, ch := range s.subs {
		select {
		case ch <- frame:
		default:
			delete(s.subs, id)
			close(ch)
		}
	}
	return frame
}

func (s *Sequencer) Resume(head int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if head > s.seq {
		s.seq = head
	}
}

func (s *Sequencer) Head() int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.seq
}

func (s *Sequencer) Watch(since *int64) ([]Frame, <-chan Frame, func()) {
	s.mu.Lock()
	defer s.mu.Unlock()
	var backlog []Frame
	if since != nil {
		backlog = make([]Frame, 0, len(s.backlog))
		for _, f := range s.backlog {
			if f.Seq > *since {
				backlog = append(backlog, f)
			}
		}
	}
	id := s.nextSub
	s.nextSub++
	ch := make(chan Frame, 4096)
	s.subs[id] = ch
	cancel := func() {
		s.mu.Lock()
		defer s.mu.Unlock()
		if existing, ok := s.subs[id]; ok {
			delete(s.subs, id)
			close(existing)
		}
	}
	return backlog, ch, cancel
}