ok

benchmark/suite_test.go

@@ -12,6 +12,25 @@ import (
 	"gfx.cafe/open/jrpc/pkg/server"
 )
 
+func TestBenchmarkSuite(t *testing.T) {
+	var executeTest = jrpctest.TestExecutor(rdwr.ServerMaker)
+	var makeTest = func(name string, fm jrpctest.TestContext) {
+		t.Run(name, func(t *testing.T) {
+			executeTest(t, fm)
+		})
+	}
+
+	ctx := context.Background()
+	makeTest("SingleClient", func(t *testing.T, server *server.Server, client codec.Conn) {
+		for i := 0; i < 10; i++ {
+			err := client.Do(ctx, nil, "test_ping", nil)
+			if err != nil {
+				t.Error(err)
+			}
+		}
+	})
+}
+
 func runBenchmarkSuite(b *testing.B, sm jrpctest.ServerMaker) {
 	ctx := context.Background()
 	executeBench := jrpctest.BenchExecutor(sm)

contrib/codecs/rdwr/codec.go

@@ -48,7 +48,8 @@ func (c *Codec) PeerInfo() codec.PeerInfo {
 func (c *Codec) decodeSingleMessage(ctx context.Context) (*serverutil.Bundle, error) {
 	c.decLock.Lock()
 	defer c.decLock.Unlock()
-	c.decBuf = c.decBuf[:0]
+	//c.decBuf = c.decBuf[:0]
+	c.decBuf = json.RawMessage{}
 	err := c.dec.DecodeContext(ctx, &c.decBuf)
 	if err != nil {
 		return nil, err

contrib/codecs/websocket/codec.go

@@ -3,7 +3,6 @@ package websocket
 import (
 	"context"
 	"io"
-	"log"
 	"net/http"
 	"sync"
 	"time"
@@ -17,12 +16,6 @@ import (
 	"gfx.cafe/open/jrpc/pkg/serverutil"
 )
 
-func init() {
-	go func() {
-		log.Println(http.ListenAndServe("localhost:6060", nil))
-	}()
-}
-
 type Codec struct {
 	closed chan struct{}
 	conn   *websocket.Conn

pkg/clientutil/idreply.go

@@ -31,6 +31,22 @@ func (i *IdReply) NextId() *codec.ID {
 	return codec.NewNumberIDPtr(i.id.Add(1))
 }
 
+func (i *IdReply) makeOrTake(id []byte) chan msgOrError {
+	i.mu.Lock()
+	defer i.mu.Unlock()
+	ch, ok := i.chs[string(id)]
+	if ok {
+		// take
+		delete(i.chs, string(id))
+	} else {
+		// make
+		ch = make(chan msgOrError, 1)
+		i.chs[string(id)] = ch
+	}
+	return ch
+
+}
+
 func (i *IdReply) make(id []byte) <-chan msgOrError {
 	i.mu.Lock()
 	defer i.mu.Unlock()
@@ -54,7 +70,7 @@ func (i *IdReply) remove(id []byte) {
 }
 
 func (i *IdReply) Resolve(id []byte, msg json.RawMessage, err error) {
-	ch := i.take(id)
+	ch := i.makeOrTake(id)
 	if ch == nil {
 		return
 	}
@@ -73,7 +89,7 @@ func (i *IdReply) Resolve(id []byte, msg json.RawMessage, err error) {
 
 func (i *IdReply) Ask(ctx context.Context, id []byte) (json.RawMessage, error) {
 	select {
-	case resp := <-i.make(id):
+	case resp := <-i.makeOrTake(id):
 		return resp.msg, resp.err
 	case <-ctx.Done():
 		i.remove(id)

pkg/server/responsewriter.go

@@ -2,10 +2,8 @@ package server
 
 import (
 	"context"
-	"log"
 	"net/http"
 	"sync"
-	"time"
 
 	"gfx.cafe/open/jrpc/pkg/codec"
 	"github.com/goccy/go-json"
@@ -63,14 +61,10 @@ func (c *callRespWriter) Send(v any, e error) (err error) {
 		}
 		return nil
 	}
-	s := time.Now()
-	log.Println("try")
 	err = c.cr.mu.Acquire(c.ctx, 1)
 	if err != nil {
 		return err
 	}
-	log.Println("release", time.Since(s))
-	s2 := time.Now()
 	defer c.cr.mu.Release(1)
 	err = c.cr.send(c.ctx, &callEnv{
 		v:           &v,
@@ -78,7 +72,6 @@ func (c *callRespWriter) Send(v any, e error) (err error) {
 		id:          c.msg.ID,
 		extrafields: c.msg.ExtraFields,
 	})
-	log.Println("release", time.Since(s2))
 	err = c.cr.remote.Flush()
 	if err != nil {
 		return err

pkg/server/server.go

@@ -68,7 +68,6 @@ func (s *Server) ServeCodec(ctx context.Context, remote codec.ReaderWriter) erro
 			}()
 		}
 	}()
-	wg.Wait()
 	allErrs = append(allErrs, err)
 	if len(allErrs) > 0 {
 		return errors.Join(allErrs...)
@@ -143,13 +142,6 @@ func (s *Server) serveBatch(ctx context.Context,
 				batchResults = append(batchResults, v)
 			}
 		}
-		// early respond to nil requests
-		if v.err != nil {
-			v.sendCalled = true
-			v.doneMu.Release(1)
-			wg.Done()
-			continue
-		}
 		// now process each request in its own goroutine
 		// TODO: stress test this.
 		go func() {
@@ -162,9 +154,7 @@ func (s *Server) serveBatch(ctx context.Context,
 			s.services.ServeRPC(v, req)
 		}()
 	}
-
 	if r.batch {
-		// we only need to do this if this is a batch call with requests
 		err = doneMu.Acquire(ctx, int64(totalRequests))
 		if err != nil {
 			return err