race

contrib/codecs/broker/broker_inproc.go

@@ -3,7 +3,6 @@ package broker
 import (
 	"context"
 	"encoding/json"
-	"log"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -66,13 +65,10 @@ func (b *ChannelBroker) SetDroppedMessageHandler(fn func(string, []byte)) *Chann
 }
 
 func (b *ChannelBroker) ReadRequest(ctx context.Context) (json.RawMessage, func(json.RawMessage) error, error) {
-	log.Println("start recv")
 	select {
 	case <-ctx.Done():
-		log.Println("doned")
 		return nil, nil, ctx.Err()
 	case f := <-b.msgs:
-		log.Println("recv", string(f.data))
 		return f.data, func(resp json.RawMessage) error {
 			return b.Publish(context.Background(), f.topic, resp)
 		}, nil
@@ -84,7 +80,6 @@ func (b *ChannelBroker) WriteRequest(ctx context.Context, topic string, msg json
 	case <-ctx.Done():
 		return ctx.Err()
 	case b.msgs <- &frame{data: msg, topic: topic}:
-		log.Println("wrote", string(msg))
 	}
 	return nil
 }

contrib/codecs/broker/server.go

@@ -25,6 +25,6 @@ func (s *Server) ServeSpoke(ctx context.Context, stream ServerSpoke) {
 			continue
 		}
 		cd := NewCodec(req, fn)
-		s.Server.ServeCodec(ctx, cd)
+		go s.Server.ServeCodec(ctx, cd)
 	}
 }

pkg/util/mapset/mapset.go

@@ -26,6 +26,8 @@ func (s *Set[T]) Remove(x T) {
 }
 
 func (s *Set[T]) Each(fn func(x T) bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	for k := range s.m {
 		if !fn(k) {
 			return