fix data races

rpc.go

@@ -33,45 +33,36 @@ func (r *rpcClient) do(ctx context.Context, method string, args, result interfac
 	return waiter.Wait(ctx, result)
 }
 
+type rpcHandlerFunc = func(*jsonrpc2.Request)
+
 type rpcHandler struct {
-	mutex         sync.Mutex
-	eventHandlers map[string][]chan *jsonrpc2.Request
+	handlersMu sync.Mutex
+	handlers   map[string][]rpcHandlerFunc
 }
 
 func newRPCHandler() *rpcHandler {
 	return &rpcHandler{
-		eventHandlers: make(map[string][]chan *jsonrpc2.Request),
+		handlers: make(map[string][]rpcHandlerFunc),
 	}
 }
 
-// TODO: document obligations around chan. It appears to be used for at most one request.
-func (r *rpcHandler) registerEventHandler(eventMethod string) <-chan *jsonrpc2.Request {
-	r.mutex.Lock()
-	defer r.mutex.Unlock()
-
-	ch := make(chan *jsonrpc2.Request)
-	r.eventHandlers[eventMethod] = append(r.eventHandlers[eventMethod], ch)
-	return ch
+// registerEventHandler registers a handler for the specified event.
+// After the next occurrence of the event, the handler will be called,
+// once, in its own goroutine.
+func (r *rpcHandler) registerEventHandler(eventMethod string, h rpcHandlerFunc) {
+	r.handlersMu.Lock()
+	r.handlers[eventMethod] = append(r.handlers[eventMethod], h)
+	r.handlersMu.Unlock()
 }
 
+// Handle calls all registered handlers for the request, concurrently, each in its own goroutine.
 func (r *rpcHandler) Handle(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.Request) {
-	r.mutex.Lock()
-	handlers := r.eventHandlers[req.Method]
-	r.eventHandlers[req.Method] = nil
-	r.mutex.Unlock()
+	r.handlersMu.Lock()
+	handlers := r.handlers[req.Method]
+	r.handlers[req.Method] = nil
+	r.handlersMu.Unlock()
 
-	if len(handlers) > 0 {
-		go func() {
-			// Broadcast the request to each handler in sequence.
-			// TODO rethink this. needs function call.
-			for _, handler := range handlers {
-				select {
-				case handler <- req:
-				case <-ctx.Done():
-					// TODO: ctx.Err
-					break
-				}
-			}
-		}()
+	for _, h := range handlers {
+		go h(req)
 	}
 }

rpc_test.go

@@ -10,7 +10,10 @@ import (
 
 func TestRPCHandlerEvents(t *testing.T) {
 	rpcHandler := newRPCHandler()
-	eventCh := rpcHandler.registerEventHandler("somethingHappened")
+	eventCh := make(chan *jsonrpc2.Request)
+	rpcHandler.registerEventHandler("somethingHappened", func(req *jsonrpc2.Request) {
+		eventCh <- req
+	})
 	go func() {
 		time.Sleep(1 * time.Second)
 		rpcHandler.Handle(context.Background(), nil, &jsonrpc2.Request{Method: "somethingHappened"})

terminal.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/sourcegraph/jsonrpc2"
 	"golang.org/x/crypto/ssh"
 )
 
@@ -71,12 +72,15 @@ func (t TerminalCommand) Run(ctx context.Context) (io.ReadCloser, error) {
 		ReadOnlyForGuests: false,
 	}
 
-	terminalStarted := t.terminal.client.rpc.handler.registerEventHandler("terminal.terminalStarted")
+	started := make(chan struct{})
+	t.terminal.client.rpc.handler.registerEventHandler("terminal.terminalStarted", func(*jsonrpc2.Request) {
+		close(started)
+	})
 	var result startTerminalResult
 	if err := t.terminal.client.rpc.do(ctx, "terminal.startTerminal", &args, &result); err != nil {
 		return nil, fmt.Errorf("error making terminal.startTerminal call: %v", err)
 	}
-	<-terminalStarted
+	<-started
 
 	channel, err := t.terminal.client.openStreamingChannel(ctx, result.StreamName, result.StreamCondition)
 	if err != nil {
@@ -101,7 +105,10 @@ func (t terminalReadCloser) Read(b []byte) (int, error) {
 }
 
 func (t terminalReadCloser) Close() error {
-	terminalStopped := t.terminalCommand.terminal.client.rpc.handler.registerEventHandler("terminal.terminalStopped")
+	stopped := make(chan struct{})
+	t.terminalCommand.terminal.client.rpc.handler.registerEventHandler("terminal.terminalStopped", func(*jsonrpc2.Request) {
+		close(stopped)
+	})
 	if err := t.terminalCommand.terminal.client.rpc.do(context.Background(), "terminal.stopTerminal", []int{t.terminalID}, nil); err != nil {
 		return fmt.Errorf("error making terminal.stopTerminal call: %v", err)
 	}
@@ -110,7 +117,7 @@ func (t terminalReadCloser) Close() error {
 		return fmt.Errorf("error closing channel: %v", err)
 	}
 
-	<-terminalStopped
+	<-stopped
 
 	return nil
 }