From 0ad9fd89958a679a22bff55ad8a7e4c694066a96 Mon Sep 17 00:00:00 2001
From: Quinn Slack <qslack@qslack.com>
Date: Tue, 31 Jan 2017 23:39:03 -0800
Subject: [PATCH] synchronize writes in BufferedObjectStream (#9)

Running various applications that use jsonrpc2 with the Go race detector shows that there is a race condition where `WriteObject` can be called from concurrent goroutines (e.g., 1 sending a request, 1 writing a response).
---
 stream.go | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/stream.go b/stream.go
index 3095592..f146f11 100644
--- a/stream.go
+++ b/stream.go
@@ -8,6 +8,7 @@ import (
 	"io"
 	"strconv"
 	"strings"
+	"sync"
 )
 
 // An ObjectStream is a bidirectional stream of JSON-RPC 2.0 objects.
@@ -30,6 +31,8 @@ type bufferedObjectStream struct {
 	r    *bufio.Reader
 
 	codec ObjectCodec
+
+	mu sync.Mutex
 }
 
 // NewBufferedStream creates a buffered stream from a network
@@ -37,7 +40,7 @@ type bufferedObjectStream struct {
 // objectStream is used to produce the bytes to write to the stream
 // for the JSON-RPC 2.0 objects.
 func NewBufferedStream(conn io.ReadWriteCloser, codec ObjectCodec) ObjectStream {
-	return bufferedObjectStream{
+	return &bufferedObjectStream{
 		conn:  conn,
 		w:     bufio.NewWriter(conn),
 		r:     bufio.NewReader(conn),
@@ -46,7 +49,9 @@ func NewBufferedStream(conn io.ReadWriteCloser, codec ObjectCodec) ObjectStream
 }
 
 // WriteObject implements ObjectStream.
-func (t bufferedObjectStream) WriteObject(obj interface{}) error {
+func (t *bufferedObjectStream) WriteObject(obj interface{}) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
 	if err := t.codec.WriteObject(t.w, obj); err != nil {
 		return err
 	}
@@ -54,12 +59,12 @@ func (t bufferedObjectStream) WriteObject(obj interface{}) error {
 }
 
 // ReadObject implements ObjectStream.
-func (t bufferedObjectStream) ReadObject(v interface{}) error {
+func (t *bufferedObjectStream) ReadObject(v interface{}) error {
 	return t.codec.ReadObject(t.r, v)
 }
 
 // Close implements ObjectStream.
-func (t bufferedObjectStream) Close() error {
+func (t *bufferedObjectStream) Close() error {
 	return t.conn.Close()
 }