new event system (again)

client/config_ignorer.go

@@ -20,9 +20,6 @@ func (c *ConfigIgnorer) OnStart() (err error) {
 }
 
 func (c *ConfigIgnorer) Default(e event.Anything) (err error) {
-	if e.Val == (event.Tick{}) {
-		return
-	}
 	slog.Debug("Config ignorer", "packet", fmt.Sprintf("%#v", e))
 	return
 }

client/login.go

@@ -46,10 +46,6 @@ func (s *LoginClient) OnDefault(event event.Unhandled) (err error) {
 	return
 }
 
-func (s *LoginClient) OnCycle(_ event.Tick) (err error) {
-	return
-}
-
 func (s *LoginClient) OnClose(_ event.Close) (err error) {
 	return
 }
@@ -133,7 +129,6 @@ func LoginRaw(c *proto.Conn, account *api.Account, addr string) (err error) {
 		Account: account,
 	}
 	lc.RegisterCritical(lc.OnDefault)
-	lc.RegisterCritical(lc.OnCycle)
 	lc.RegisterCritical(lc.OnClose)
 	lc.RegisterUntilLatest(lc.OnCompress)
 	lc.RegisterUntilLatest(lc.OnEncrypt)

event/event_bus.go

@@ -7,20 +7,14 @@ import (
 	"fmt"
 	"log/slog"
 	"reflect"
-	"time"
-)
-
-const (
-	LoopsPerSecond = 10000 // this is just so the cpu doesn't burn itself in an unlimited infinite loop
-	TimePerLoop    = time.Second / LoopsPerSecond
 )
 
 type Loop struct {
 	HandlerFunctions map[reflect.Type][]reflect.Value
 
-	Ctx     context.Context
+	Ctx          context.Context
-	ErrChan chan error
+	cancel       context.CancelFunc
-	cancel  context.CancelFunc
+	eventSources []reflect.SelectCase
 }
 
 type Unhandled struct {
@@ -31,10 +25,6 @@ type Anything struct {
 	Val any
 }
 
-type (
-	Tick struct{}
-)
-
 // Close is fired when the loop shuts down.
 type Close struct {
 	Reason error
@@ -45,6 +35,14 @@ type HandlerDoneErr struct {
 	Return error
 }
 
+// CallFunc is a special value that when fired with Loop.Fire it calls a function on the main loop
+// Its usefull for event event sources to call functions on the main loop
+// because event sources dont run on the main loop
+type CallFunc struct {
+	F    reflect.Value
+	Args []reflect.Value
+}
+
 func (e HandlerDoneErr) Error() string {
 	return "HandlerDoneErr"
 }
@@ -73,6 +71,7 @@ var WhileClosingErr = errors.New("error while closing")
 // CloseNonCriticalErr will close even when the event is non-critical
 var CloseNonCriticalErr = errors.New("close non-critical")
 var ContextDoneErr = errors.New("context done")
+var RecvNotOkErr = errors.New("recv not ok")
 
 func ValidateHandler(h any) (eventType reflect.Type, hv reflect.Value) {
 	hv = reflect.ValueOf(h)
@@ -89,6 +88,29 @@ func ValidateHandler(h any) (eventType reflect.Type, hv reflect.Value) {
 	return
 }
 
+// RegisterEventSource registers an event source an event source is a channel this channel along with all other event sources will be reflect.Select 'ed until the loop closes
+// here is an example of an event source function
+//
+//	func() (chSending <-chan any) {
+//			ch := make(chan any)
+//			go func() {
+//				ctx := conn.Ctx
+//				for {
+//					select {
+//					case <-ctx.Done():
+//						return
+//					case eventt := <-otherSource:
+//						err = c.DoCommand(eventt)
+//					}
+//				}
+//			}()
+//			chSending = ch
+//			return
+//		}
+func (l *Loop) RegisterEventSource(ch <-chan any) {
+	l.eventSources = append(l.eventSources, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)})
+}
+
 // Register registers a handler that will not close the loop if it returns an error.
 // Instead, it will log it to slog.Error
 // Warning: If a bad handler is passed in the program will panic.
@@ -137,7 +159,7 @@ func (l *Loop) FireFound(event any) (found bool, err error) {
 			errV := handler.Call([]reflect.Value{reflect.ValueOf(event)})[0]
 			if !errV.IsNil() {
 				err = errV.Interface().(error)
-				err = errors.Join(fmt.Errorf("event handler failed at: %s(%T)", handler.Type().In(0), event), err)
+				err = errors.Join(fmt.Errorf("event handler failed at: %T", event), err)
 				return
 			}
 		}
@@ -147,6 +169,21 @@ func (l *Loop) FireFound(event any) (found bool, err error) {
 
 // Fire dispatches the provided event to all appropriate handlers and invokes fallback handlers if the event is unhandled.
 func (l *Loop) Fire(event any) (err error) {
+	err, isErr := event.(error)
+	if isErr {
+		return
+	}
+	err = nil
+
+	f, isCallFunc := event.(CallFunc)
+	if isCallFunc {
+		errV := f.F.Call(f.Args)[0].Interface()
+		if errV != nil {
+			err = errV.(error)
+		}
+		return
+	}
+
 	found, err := l.FireFound(event)
 	if err != nil {
 		if errors.Is(err, DontForwardErr) {
@@ -164,24 +201,30 @@ func (l *Loop) Fire(event any) (err error) {
 	return
 }
 
-func (l *Loop) OnTick(_ Tick) (err error) {
+func (l *Loop) ContextDoneListener() (chSending <-chan any) {
-	select {
+	ch := make(chan any)
-	case <-l.Ctx.Done():
+	go func() {
-		err = ContextDoneErr
+		select {
-	default:
+		case <-l.Ctx.Done():
-	}
+			ch <- ContextDoneErr
-	return
+		}
+	}()
+	return ch
 }
 
-func (l *Loop) startLoop() (err error) {
+func (l *Loop) StartLoop() (err error) {
 	_ = *l // exit early on nil loop
+	l.RegisterEventSource(l.ContextDoneListener())
 	for {
-		tickStartTime := time.Now()
+		_, v, ok := reflect.Select(l.eventSources)
-		err = l.Fire(Tick{})
+		if !ok {
+			err = RecvNotOkErr
+			break
+		}
+		err = l.Fire(v.Interface())
 		if err != nil {
 			break
 		}
-		time.Sleep(TimePerLoop - time.Since(tickStartTime))
 	}
 	if errors.Is(err, HandlerDoneErr{}) {
 		err = errors.Unwrap(err)
@@ -193,25 +236,8 @@ func (l *Loop) startLoop() (err error) {
 	return
 }
 
-func (l *Loop) StartLoop() (err error) {
-	go func() {
-		l.ErrChan <- MainHandlerErr{l.startLoop()}
-	}()
-	err = <-l.ErrChan
-	l.cancel()
-	errMaybe := MainHandlerErr{}
-	if errors.As(err, &errMaybe) {
-		err = errors.Unwrap(err)
-	} else {
-		for !errors.Is(<-l.ErrChan, MainHandlerErr{}) {
-		}
-	}
-	return
-}
-
 func NewLoop() (l *Loop) {
 	l = &Loop{}
-	l.ErrChan = make(chan error)
 	l.Ctx, l.cancel = context.WithCancel(context.Background())
 	l.HandlerFunctions = make(map[reflect.Type][]reflect.Value)
 	return

event/stdin.go

@@ -2,7 +2,6 @@ package event
 
 import (
 	"bufio"
-	"context"
 	"os"
 )
 
@@ -10,36 +9,26 @@ type Stdin struct {
 	Val string
 }
 
-type StdinHandler struct {
+func StartListingStdin(l *Loop) {
-	*Loop
+	l.RegisterEventSource(func() (chSending <-chan any) {
-	stdInChan chan Stdin
+		ch := make(chan any)
-	Ctx       context.Context
+		go func() {
-}
+			scanner := bufio.NewScanner(os.Stdin)
-
+			ctx := l.Ctx
-func (s *StdinHandler) OnTick(_ Tick) (err error) {
+			for scanner.Scan() {
-	select {
+				select {
-	case stdin := <-s.stdInChan:
+				case <-ctx.Done():
-		err = s.Fire(Stdin{stdin.Val})
+					return
-	default:
+				default:
-
+					ch <- Stdin{scanner.Text()}
-	}
+				}
-	return
+			}
-}
+			if err := scanner.Err(); err != nil {
-
+				ch <- err
-func (s *StdinHandler) ScanJob() {
+			}
-	scanner := bufio.NewScanner(os.Stdin)
+		}()
-	for scanner.Scan() {
+		chSending = ch
-		select {
+		return
-		case <-s.Ctx.Done():
+	}())
-			return
-		default:
-			s.stdInChan <- Stdin{scanner.Text()}
-		}
-	}
-}
-func (s *StdinHandler) OnStart() (err error) {
-	s.RegisterCritical(s.OnTick)
-	s.stdInChan = make(chan Stdin)
-	go s.ScanJob()
 	return
 }

event/timer.go

@@ -1,6 +1,8 @@
 package event
 
 import (
+	"context"
+	"reflect"
 	"slices"
 	"time"
 )
@@ -48,6 +50,44 @@ func (t *Timer) Tick() (err error) {
 	return
 }
 
+func (t *Timer) Trigger(ch chan any) {
+	triggerAt := time.Time{}
+	found := false
+	iOnce, iInterval := -1, -1
+	var oTask OnceTask
+	for iOnce, oTask = range t.Once {
+		if oTask.Trigger.Before(triggerAt) || triggerAt.IsZero() {
+			found = true
+			triggerAt = oTask.Trigger
+			break
+		}
+	}
+	var iTask IntervalTask
+	for iInterval, iTask = range t.Interval {
+		if iTask.LastTrigger.Add(iTask.Interval).Before(triggerAt) || triggerAt.IsZero() {
+			found = true
+			triggerAt = iTask.LastTrigger.Add(iTask.Interval)
+			break
+		}
+	}
+	if !found {
+		return
+	}
+	now := time.Now()
+	time.Sleep(triggerAt.Sub(now))
+	if iOnce >= 0 {
+		ch <- CallFunc{F: reflect.ValueOf(t.Once[iOnce].F), Args: []reflect.Value{}}
+		t.Once = slices.Delete(t.Once, iOnce, iOnce+1)
+		return
+	}
+	if iInterval >= 0 {
+		ch <- CallFunc{F: reflect.ValueOf(t.Interval[iInterval].F), Args: []reflect.Value{}}
+		t.Interval[iInterval].LastTrigger = now
+		return
+	}
+	panic("unreachable")
+}
+
 func (t *Timer) Every(interval time.Duration, f func() error) {
 	t.Interval = append(t.Interval, IntervalTask{
 		F:        f,
@@ -61,3 +101,23 @@ func (t *Timer) In(interval time.Duration, f func() error) {
 		Trigger: time.Now().Add(interval),
 	})
 }
+
+func (t *Timer) MakeEventSource(ctx context.Context) (chSending <-chan any) {
+	ch := make(chan any)
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				t.Trigger(ch)
+			}
+		}
+	}()
+	chSending = ch
+	return
+}
+
+func (t *Timer) Start(c *Loop) {
+	c.RegisterEventSource(t.MakeEventSource(c.Ctx))
+}

proto/connection.go

@@ -2,7 +2,6 @@ package proto
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -16,6 +15,10 @@ import (
 	"github.com/admin-else/strom/proto_generated"
 )
 
+const (
+	BufferNPackets = 100
+)
+
 var (
 	BadPacketTypeErr                    = errors.New("bad packet type")
 	BadPacketIdErr                      = errors.New("bad packet id")
@@ -26,6 +29,9 @@ var (
 	PacketDoesntExistInFutureVersionErr = errors.New("packet doesnt exist in future version")
 )
 
+// LoginTick is used as a subsite for the old event system because it is very useful in login contexts
+type LoginTick struct{}
+
 var LatestVersion = proto_generated.SupportedVersions[len(proto_generated.SupportedVersions)-1]
 
 // UnCodablePacket represents a packet that could not be decoded.
@@ -54,35 +60,28 @@ type Conn struct {
 	Version         string
 	ProtocolVersion int32
 
-	receiveRoutine bool
+	sendTickAlways bool
-	packetCh       chan proto_base.EncodeDecodeAble
 }
 
 func (c *Conn) StartConn() (err error) {
-	c.RegisterCritical(c.OnTick)
+	if c.State() != proto_base.Play {
-	if c.state == proto_base.Play {
+		c.sendTickAlways = true
-		c.ActivateReceiveRoutine()
 	}
+
+	c.RegisterEventSource(c.ActivateReceiveRoutine())
+	c.RegisterCritical(c.OnLoginTick)
 	err = c.Loop.StartLoop()
 	c.Loop = event.NewLoop()
 	return
 }
 
-func (c *Conn) ReceiveJob() {
+func (c *Conn) OnLoginTick(_ LoginTick) (err error) {
-	ctx := c.Ctx
+	packet, err := c.Receive()
-	for {
+	if err != nil {
-		select {
+		return
-		case <-ctx.Done():
-			return
-		default:
-			packet, err := c.Receive()
-			if err != nil {
-				c.ErrChan <- err
-				return
-			}
-			c.packetCh <- packet
-		}
 	}
+	err = c.Fire(packet)
+	return
 }
 
 func (c *Conn) SetState(state proto_base.State) {
@@ -119,13 +118,30 @@ func (c *Conn) SetProtocolVersion(version int32) (err error) {
 
 // ActivateReceiveRoutine activates the receive-routine.
 // It ensures that the receive-routine is only started once and creates a buffered channel for packet handling.
-func (c *Conn) ActivateReceiveRoutine() {
+func (c *Conn) ActivateReceiveRoutine() (chSending <-chan any) {
-	if c.receiveRoutine {
+	ch := make(chan any, BufferNPackets)
-		return
+	go func() {
-	}
+		ctx := c.Ctx // this prevents c.Ctx being reassigned
-	c.packetCh = make(chan proto_base.EncodeDecodeAble, 100) // i dunno how many to buffer 100 seems reasonable
+		for {
-	go c.ReceiveJob()
+			select {
-	c.receiveRoutine = true
+			case <-ctx.Done():
+				return
+			default:
+				if c.sendTickAlways {
+					ch <- LoginTick{}
+				} else {
+					packet, err := c.Receive()
+					if err != nil {
+						ch <- err
+						return
+					}
+					ch <- packet
+				}
+			}
+		}
+	}()
+	chSending = ch
+	return
 }
 
 func (c *Conn) translatePacketVersion(packet proto_base.EncodeDecodeAble, packetInfo proto_base.PacketInfo) (convertedPacket proto_base.EncodeDecodeAble, err error) {
@@ -216,25 +232,3 @@ func (c *Conn) Receive() (packet proto_base.EncodeDecodeAble, err error) {
 	}
 	return
 }
-
-func (c *Conn) OnTick(_ event.Tick) (err error) {
-	var packet proto_base.EncodeDecodeAble
-	if c.receiveRoutine {
-		select {
-		case packet = <-c.packetCh:
-			if slog.Default().Enabled(context.Background(), slog.LevelDebug) {
-				slog.Debug("receive", "actor", c.Actor, "packet", fmt.Sprintf("%#v", packet))
-			}
-			err = c.Loop.Fire(packet)
-		default:
-		}
-	} else {
-		packet, err = c.Receive()
-		if err != nil {
-			return
-		}
-		slog.Debug("receive", "actor", c.Actor, "packet", fmt.Sprintf("%#v", packet))
-		err = c.Loop.Fire(packet)
-	}
-	return
-}