Repository: creachadair/jrpc2
Pull request #14 (2020-03-18 22:48:24)

A new interface for server.Loop.
Based on the comment thread discussion from #13.

creachadair: > I gave it a spin and I like it.

That's good to hear, thanks for taking a look. I think I'm mostly convinced that the caching wasn't really necessary, so I'll polish this up and write some more tests, and then merge it.

Re: SimpleServer. Let me think about that some more. It seems like something that could maybe fit into the `server` package. I'll try out a prototype and see how it feels.

Patch:
diff --git a/cmd/examples/jsh/jsh.go b/cmd/examples/jsh/jsh.go
index dafc124..3c42abb 100644
--- a/cmd/examples/jsh/jsh.go
+++ b/cmd/examples/jsh/jsh.go
@@ -99,9 +99,9 @@ func main() {
 	}
 	log.Printf("Listening for connections at %s...", lst.Addr())
 
-	server.Loop(lst, handler.Map{
+	server.Loop(lst, server.NewStatic(handler.Map{
 		"Run": handler.New(Run),
-	}, &server.LoopOptions{
+	}), &server.LoopOptions{
 		ServerOptions: &jrpc2.ServerOptions{
 			AllowV1:       true,
 			Logger:        lw,
diff --git a/cmd/examples/server/server.go b/cmd/examples/server/server.go
index fa5296f..b0fa281 100644
--- a/cmd/examples/server/server.go
+++ b/cmd/examples/server/server.go
@@ -97,7 +97,7 @@ func main() {
 		log.Fatalln("Listen:", err)
 	}
 	log.Printf("Listening at %v...", lst.Addr())
-	server.Loop(lst, mux, &server.LoopOptions{
+	server.Loop(lst, server.NewStatic(mux), &server.LoopOptions{
 		ServerOptions: &jrpc2.ServerOptions{
 			Logger:      log.New(os.Stderr, "[jrpc2.Server] ", log.LstdFlags|log.Lshortfile),
 			Concurrency: *maxTasks,
diff --git a/server/loop.go b/server/loop.go
index 9f128ba..5e02afb 100644
--- a/server/loop.go
+++ b/server/loop.go
@@ -9,16 +9,34 @@ import (
 	"github.com/creachadair/jrpc2/channel"
 )
 
+// Service is the interface used by the Loop function to start up a server.
+type Service interface {
+	// This method is called to create an assigner and initialize the service
+	// for use.  If it reports an error, the server is not started.
+	Assigner() (jrpc2.Assigner, error)
+
+	// This method is called when the server for this service has exited.
+	Finish(jrpc2.ServerStatus)
+}
+
+type singleton struct{ assigner jrpc2.Assigner }
+
+func (s singleton) Assigner() (jrpc2.Assigner, error) { return s.assigner, nil }
+func (singleton) Finish(jrpc2.ServerStatus)           {}
+
+// NewStatic creates a static (singleton) service from the given assigner.
+func NewStatic(assigner jrpc2.Assigner) func() Service {
+	svc := singleton{assigner}
+	return func() Service { return svc }
+}
+
 // Loop obtains connections from lst and starts a server for each with the
-// given assigner and options, running in a new goroutine. If accept reports an
-// error, the loop will terminate and the error will be reported once all the
-// servers currently active have returned.
-//
-// While running, Loop maintains a pool of *jrpc2.Server values to reduce setup
-// and memory overhead. However, it does not rate-limit connections.
+// given service constructor and options, running in a new goroutine. If accept
+// reports an error, the loop will terminate and the error will be reported
+// once all the servers currently active have returned.
 //
 // TODO: Add options to support sensible rate-limitation.
-func Loop(lst net.Listener, assigner jrpc2.Assigner, opts *LoopOptions) error {
+func Loop(lst net.Listener, newService func() Service, opts *LoopOptions) error {
 	newChannel := opts.framing()
 	serverOpts := opts.serverOpts()
 	log := func(string, ...interface{}) {}
@@ -26,11 +44,6 @@ func Loop(lst net.Listener, assigner jrpc2.Assigner, opts *LoopOptions) error {
 		log = serverOpts.Logger.Printf
 	}
 
-	// Maintain a pool of servers to save the server setup cost.
-	pool := &sync.Pool{
-		New: func() interface{} { return jrpc2.NewServer(assigner, serverOpts) },
-	}
-
 	var wg sync.WaitGroup
 	for {
 		conn, err := lst.Accept()
@@ -47,10 +60,17 @@ func Loop(lst net.Listener, assigner jrpc2.Assigner, opts *LoopOptions) error {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			srv := pool.Get().(*jrpc2.Server).Start(ch)
-			defer pool.Put(srv)
-			if err := srv.Wait(); err != nil {
-				log("Server exit: %v", err)
+			svc := newService()
+			assigner, err := svc.Assigner()
+			if err != nil {
+				log("Service initialization failed: %v", err)
+				return
+			}
+			srv := jrpc2.NewServer(assigner, serverOpts).Start(ch)
+			stat := srv.WaitStatus()
+			svc.Finish(stat)
+			if stat.Err != nil {
+				log("Server exit: %v", stat.Err)
 			}
 		}()
 	}


Test patch:
diff --git a/server/loop_test.go b/server/loop_test.go
index 5e0cc09..8489096 100644
--- a/server/loop_test.go
+++ b/server/loop_test.go
@@ -15,6 +15,50 @@ import (
 
 var newChan = channel.Varint
 
+// A static test service that returns the same thing each time.
+var testService = NewStatic(handler.Map{
+	"Test": handler.New(func(context.Context) (string, error) {
+		return "OK", nil
+	}),
+})
+
+// A service with session state, to exercise start/finish plumbing.
+type testSession struct {
+	t     *testing.T
+	init  bool
+	nCall int
+}
+
+func newTestSession(t *testing.T) func() Service {
+	return func() Service { t.Helper(); return &testSession{t: t} }
+}
+
+func (t *testSession) Assigner() (jrpc2.Assigner, error) {
+	if t.init {
+		t.t.Error("Service has already been initalized")
+	}
+	t.init = true
+	return handler.Map{
+		"Test": handler.New(func(context.Context) (string, error) {
+			if !t.init {
+				t.t.Error("Handler called before service initialized")
+			}
+			t.nCall++
+			return "OK", nil
+		}),
+	}, nil
+}
+
+func (t *testSession) Finish(stat jrpc2.ServerStatus) {
+	if !t.init {
+		t.t.Error("Service finished without being initialized")
+	}
+	if !stat.Success() {
+		t.t.Errorf("Finish unsuccessful: %v", stat.Err)
+	}
+	t.t.Logf("Server finished after %d calls, err=%v", t.nCall, stat.Err)
+}
+
 func mustListen(t *testing.T) net.Listener {
 	t.Helper()
 
@@ -35,7 +79,7 @@ func mustDial(t *testing.T, addr string) *jrpc2.Client {
 	return jrpc2.NewClient(newChan(conn, conn), nil)
 }
 
-func mustServe(t *testing.T, lst net.Listener) <-chan struct{} {
+func mustServe(t *testing.T, lst net.Listener, newService func() Service) <-chan struct{} {
 	t.Helper()
 
 	sc := make(chan struct{})
@@ -43,12 +87,7 @@ func mustServe(t *testing.T, lst net.Listener) <-chan struct{} {
 		defer close(sc)
 		// Start a server loop to accept connections from the clients. This should
 		// exit cleanly once all the clients have finished and the listener closes.
-		service := handler.Map{
-			"Test": handler.New(func(context.Context) (string, error) {
-				return "OK", nil
-			}),
-		}
-		if err := Loop(lst, service, &LoopOptions{
+		if err := Loop(lst, newService, &LoopOptions{
 			Framing: newChan,
 		}); err != nil {
 			t.Errorf("Loop: unexpected failure: %v", err)
@@ -61,7 +100,7 @@ func mustServe(t *testing.T, lst net.Listener) <-chan struct{} {
 func TestSeq(t *testing.T) {
 	lst := mustListen(t)
 	addr := lst.Addr().String()
-	sc := mustServe(t, lst)
+	sc := mustServe(t, lst, testService)
 
 	for i := 0; i < 5; i++ {
 		cli := mustDial(t, addr)
@@ -79,38 +118,52 @@ func TestSeq(t *testing.T) {
 
 // Test that concurrent clients against the same server work sanely.
 func TestLoop(t *testing.T) {
-	lst := mustListen(t)
-	defer lst.Close()
-	addr := lst.Addr().String()
-	sc := mustServe(t, lst)
+	tests := []struct {
+		desc string
+		cons func() Service
+	}{
+		{"StaticService", testService},
+		{"SessionStateService", newTestSession(t)},
+	}
+	const numClients = 5
+	const numCalls = 5
 
-	// Start a bunch of clients, each of which will dial the server and make
-	// some calls at random intervals to tickle the race detector.
-	var wg sync.WaitGroup
-	for i := 0; i < 5; i++ {
-		wg.Add(1)
-		i := i
-		go func() {
-			defer wg.Done()
-			cli := mustDial(t, addr)
-			defer cli.Close()
-
-			for j := 0; j < 5; j++ {
-				time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
-				var rsp string
-				if err := cli.CallResult(context.Background(), "Test", nil, &rsp); err != nil {
-					t.Errorf("[client %d]: Test call %d: unexpected error: %v", i, j+1, err)
-				} else if rsp != "OK" {
-					t.Errorf("[client %d]: Test call %d: got %q, want OK", i, j+1, rsp)
-				}
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			lst := mustListen(t)
+			addr := lst.Addr().String()
+			sc := mustServe(t, lst, test.cons)
+
+			// Start a bunch of clients, each of which will dial the server and make
+			// some calls at random intervals to tickle the race detector.
+			var wg sync.WaitGroup
+			for i := 0; i < numClients; i++ {
+				wg.Add(1)
+				i := i
+				go func() {
+					defer wg.Done()
+					cli := mustDial(t, addr)
+					defer cli.Close()
+
+					for j := 0; j < numCalls; j++ {
+						time.Sleep(time.Duration(rand.Intn(10)) * time.Millisecond)
+						var rsp string
+						if err := cli.CallResult(context.Background(), "Test", nil, &rsp); err != nil {
+							t.Errorf("[client %d]: Test call %d: unexpected error: %v", i, j+1, err)
+						} else if rsp != "OK" {
+							t.Errorf("[client %d]: Test call %d: got %q, want OK", i, j+1, rsp)
+						}
+					}
+				}()
 			}
-		}()
-	}
 
-	// Wait for the clients to be finished and then close the listener so that
-	// the service loop will stop.
-	wg.Wait()
-	t.Log("Clients are finished; closing listener")
-	lst.Close()
-	<-sc
+			// Wait for the clients to be finished and then close the listener so that
+			// the service loop will stop.
+			wg.Wait()
+			t.Log("Clients are finished; stopping listener")
+			lst.Close()
+			<-sc
+			t.Log("Server is finished")
+		})
+	}
 }