diff --git a/client/config_ignorer.go b/client/config_ignorer.go index 8beafbb..0f7bfb8 100644 --- a/client/config_ignorer.go +++ b/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 } diff --git a/client/login.go b/client/login.go index bde3521..962d4d9 100644 --- a/client/login.go +++ b/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) diff --git a/event/event_bus.go b/event/event_bus.go index 918208a..ce45db9 100644 --- a/event/event_bus.go +++ b/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 - ErrChan chan error - cancel context.CancelFunc + Ctx context.Context + 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) { - select { - case <-l.Ctx.Done(): - err = ContextDoneErr - default: - } - return +func (l *Loop) ContextDoneListener() (chSending <-chan any) { + ch := make(chan any) + go func() { + select { + case <-l.Ctx.Done(): + ch <- ContextDoneErr + } + }() + 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() - err = l.Fire(Tick{}) + _, v, ok := reflect.Select(l.eventSources) + 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 diff --git a/event/stdin.go b/event/stdin.go index 2123faf..368c7a4 100644 --- a/event/stdin.go +++ b/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 { - *Loop - stdInChan chan Stdin - Ctx context.Context -} - -func (s *StdinHandler) OnTick(_ Tick) (err error) { - select { - case stdin := <-s.stdInChan: - err = s.Fire(Stdin{stdin.Val}) - default: - - } - return -} - -func (s *StdinHandler) ScanJob() { - scanner := bufio.NewScanner(os.Stdin) - for scanner.Scan() { - select { - 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() +func StartListingStdin(l *Loop) { + l.RegisterEventSource(func() (chSending <-chan any) { + ch := make(chan any) + go func() { + scanner := bufio.NewScanner(os.Stdin) + ctx := l.Ctx + for scanner.Scan() { + select { + case <-ctx.Done(): + return + default: + ch <- Stdin{scanner.Text()} + } + } + if err := scanner.Err(); err != nil { + ch <- err + } + }() + chSending = ch + return + }()) return } diff --git a/event/timer.go b/event/timer.go index 4062f7b..ff6a5cb 100644 --- a/event/timer.go +++ b/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)) +} diff --git a/proto/connection.go b/proto/connection.go index ae27272..34385fb 100644 --- a/proto/connection.go +++ b/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 - packetCh chan proto_base.EncodeDecodeAble + sendTickAlways bool } func (c *Conn) StartConn() (err error) { - c.RegisterCritical(c.OnTick) - if c.state == proto_base.Play { - c.ActivateReceiveRoutine() + if c.State() != proto_base.Play { + c.sendTickAlways = true } + + c.RegisterEventSource(c.ActivateReceiveRoutine()) + c.RegisterCritical(c.OnLoginTick) err = c.Loop.StartLoop() c.Loop = event.NewLoop() return } -func (c *Conn) ReceiveJob() { - ctx := c.Ctx - for { - select { - case <-ctx.Done(): - return - default: - packet, err := c.Receive() - if err != nil { - c.ErrChan <- err - return - } - c.packetCh <- packet - } +func (c *Conn) OnLoginTick(_ LoginTick) (err error) { + packet, err := c.Receive() + if err != nil { + return } + 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() { - if c.receiveRoutine { - return - } - c.packetCh = make(chan proto_base.EncodeDecodeAble, 100) // i dunno how many to buffer 100 seems reasonable - go c.ReceiveJob() - c.receiveRoutine = true +func (c *Conn) ActivateReceiveRoutine() (chSending <-chan any) { + ch := make(chan any, BufferNPackets) + go func() { + ctx := c.Ctx // this prevents c.Ctx being reassigned + for { + select { + 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 -}