-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
radio: move OneOff to the util package
OneOff now has a bigger brother named StreamValue that keeps a stream open in the background and continuously updates the value until the context used to create it is canceled. StreamValue also supports passing it a callback that is to be called everytime a value arrives. ircbot: move to the StreamValue for status updates ircbot/AnnounceSong: now checks if a song is equal to the previous one instead of leaving this up to the caller.
- Loading branch information
Showing
9 changed files
with
135 additions
and
122 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,8 +1,89 @@ | ||
| package util | ||
|
|
||
| import "net/http" | ||
| import ( | ||
| "context" | ||
| "net/http" | ||
| "sync/atomic" | ||
| "time" | ||
|
|
||
| "github.com/R-a-dio/valkyrie/util/eventstream" | ||
| "github.com/rs/zerolog" | ||
| ) | ||
|
|
||
| // IsHTMX checks if a request was made by HTMX through the Hx-Request header | ||
| func IsHTMX(r *http.Request) bool { | ||
| return r.Header.Get("Hx-Request") == "true" | ||
| } | ||
|
|
||
| type StreamFn[T any] func(context.Context) (eventstream.Stream[T], error) | ||
|
|
||
| type StreamCallbackFn[T any] func(context.Context, T) | ||
|
|
||
| // OneOff creates a stream through fn and returns the first value received after which | ||
| // it closes the stream. Should be used where you only need a very sporadic value that is | ||
| // supplied by a streaming API. | ||
| func OneOff[T any](ctx context.Context, fn StreamFn[T]) (T, error) { | ||
| s, err := fn(ctx) | ||
| if err != nil { | ||
| return *new(T), err | ||
| } | ||
| defer s.Close() | ||
|
|
||
| return s.Next() | ||
| } | ||
|
|
||
| // StreamValue opens the stream created by StreamFn and calls any callbackFn given everytime a new | ||
| // value is returned by the stream. StreamValue also stores the last received value, accessable by | ||
| // calling .Latest | ||
| func StreamValue[T any](ctx context.Context, fn StreamFn[T], callbackFn ...StreamCallbackFn[T]) *Value[T] { | ||
| var value Value[T] | ||
|
|
||
| go func() { | ||
| for { | ||
| stream, err := fn(ctx) | ||
| if err != nil { | ||
| // stream creation error most likely means the service | ||
| // is down or unavailable for some reason so retry in | ||
| // a little bit and stay alive | ||
| zerolog.Ctx(ctx).Error().Err(err).Msg("stream-value: stream error") | ||
| select { | ||
| case <-ctx.Done(): | ||
| // context was canceled, either while we were waiting on | ||
| // retrying, or that was our original error and we exit | ||
| return | ||
| case <-time.After(time.Second): | ||
| } | ||
| continue | ||
| } | ||
|
|
||
| for { | ||
| v, err := stream.Next() | ||
| if err != nil { | ||
| // we either got context canceled or received some | ||
| // stream error that indicates we need a new stream, | ||
| // try and get one from the outer loop. | ||
| zerolog.Ctx(ctx).Error().Err(err).Msg("stream-value: next error") | ||
| break | ||
| } | ||
|
|
||
| value.last.Store(&v) | ||
|
|
||
| for _, callback := range callbackFn { | ||
| // TODO(wessie): run in goroutine? | ||
| callback(ctx, v) | ||
| } | ||
| } | ||
| stream.Close() | ||
| } | ||
| }() | ||
|
|
||
| return &value | ||
| } | ||
|
|
||
| type Value[T any] struct { | ||
| last atomic.Pointer[T] | ||
| } | ||
|
|
||
| func (v *Value[T]) Latest() T { | ||
| return *v.last.Load() | ||
| } |
Oops, something went wrong.