kgo: patch HookFetchRecordUnbuffered
Previously, if a hook implemented both HookFetchRecordBuffered and
HookFetchRecordUnbuffered, then HookFetchRecordUnbuffered would never
be called. The type switch in this logic would always match the buffered
case and never call unbuffered.
Reorganizing the conditionals a little bit ensures that the correct hook
is always called.
@@ -289,9 +289,9 @@ type bufferedFetch struct {
func (s *source) hook(f *Fetch, buffered, polled bool) {
s.cl.cfg.hooks.each(func(h Hook) {
- switch h := h.(type) {
- case HookFetchRecordBuffered:
- if !buffered {
+ if buffered {
+ h, ok := h.(HookFetchRecordBuffered)
+ if !ok {
return
}
for i := range f.Topics {
@@ -303,9 +303,9 @@ func (s *source) hook(f *Fetch, buffered, polled bool) {
}
}
}
-
- case HookFetchRecordUnbuffered:
- if buffered {
+ } else {
+ h, ok := h.(HookFetchRecordUnbuffered)
+ if !ok {
return
}
for i := range f.Topics {
