Go kosu/rpc

packages/go-kosu/go.sum

@@ -207,6 +207,7 @@ github.com/tendermint/iavl v0.12.2 h1:Ls5p5VINCM1HRT9g5Vvs2zmDOCU/CCIvIHzd/pZ8P0
 github.com/tendermint/iavl v0.12.2/go.mod h1:EoKMMv++tDOL5qKKVnoIqtVPshRrEPeJ0WsgDOLAauM=
 github.com/tendermint/tendermint v0.31.5 h1:vTet8tCq3B9/J9Yo11dNZ8pOB7NtSy++bVSfkP4KzR4=
 github.com/tendermint/tendermint v0.31.5/go.mod h1:ymcPyWblXCplCPQjbOYbrF1fWnpslATMVqiGgWbZrlc=
+github.com/tendermint/tendermint v0.32.1 h1:J8ddXMbCmG6GZjdCl/N1wgdXDU9uO91J2Y5CA9xYfGo=
 github.com/tidwall/gjson v1.3.0 h1:kfpsw1W3trbg4Xm6doUtqSl9+LhLB6qJ9PkltVAQZYs=
 github.com/tidwall/gjson v1.3.0/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
 github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=

packages/go-kosu/rpc/cmd.go

@@ -30,7 +30,7 @@ func NewCommand() *cobra.Command {
 		Short: "starts the rpc bridge",
 		Long:  "The RPC bridge exposes a set of kosud functionalities over JSON-RPC 2.0",
 		PreRunE: func(cmd *cobra.Command, args []string) error {
-			if http == false && ws == false {
+			if !http && !ws {
 				return errors.New("both `--ws` and `--http` where false, you need to enable at least one")
 			}
 
@@ -54,7 +54,7 @@ func NewCommand() *cobra.Command {
 			srv := NewServer(client)
 
 			wg := sync.WaitGroup{}
-			if http == true {
+			if http {
 				wg.Add(1)
 				go func() {
 					defer wg.Done()
@@ -64,7 +64,7 @@ func NewCommand() *cobra.Command {
 				}()
 			}
 
-			if ws == true {
+			if ws {
 				wg.Add(1)
 				go func() {
 					defer wg.Done()

packages/go-kosu/rpc/doctool/main.go

@@ -10,6 +10,7 @@ import (
 	"os"
 )
 
+// nolint
 type DocEntry struct {
 	Method string `json:"method"`
 	Text   string `json:"text"`

packages/go-kosu/rpc/rpc.go

@@ -6,8 +6,11 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
+// NewServer returns a new Server which holds the registered rpc service
 func NewServer(abci *abci.Client) *rpc.Server {
 	srv := rpc.NewServer()
-	srv.RegisterName("kosu", &Service{abci: abci})
+	if err := srv.RegisterName("kosu", &Service{abci: abci}); err != nil {
+		panic(err)
+	}
 	return srv
 }

packages/go-kosu/rpc/rpc_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -17,29 +18,40 @@ import (
 func TestRPCLatestHeight(t *testing.T) {
 	_, closer := tests.StartServer(t, db.NewMemDB())
 	defer closer()
-	client := DialInProc(
+	client := rpc.DialInProc(
 		NewServer(
 			abci.NewHTTPClient("http://localhost:26657", nil),
 		),
 	)
 
+	var latest uint64
 	// Get the initial (prior the first block is mined)
-	latest, err := client.LatestHeight()
-	require.NoError(t, err)
+	require.NoError(t, client.Call(&latest, "kosu_latestHeight"))
 	assert.EqualValues(t, 0, latest)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-	fn := func(i interface{}) {
+	fn := func(_ interface{}) {
 		// this is invoked when a block is mined
-		latest, err := client.LatestHeight()
-		require.NoError(t, err)
+		require.NoError(t, client.Call(&latest, "kosu_latestHeight"))
 		assert.EqualValues(t, 1, latest)
-
 		cancel()
 	}
 
-	err = client.Subscribe(ctx, fn, "tm.event = 'NewBlock'")
+	ch := make(chan interface{})
+	defer close(ch)
+
+	sub, err := client.Subscribe(ctx, "kosu", ch, "subscribe", "tm.event = 'NewBlock'")
+	defer sub.Unsubscribe()
 	require.NoError(t, err)
 
-	<-ctx.Done()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case err := <-sub.Err():
+			t.Error(err)
+		case e := <-ch:
+			fn(e)
+		}
+	}
 }

packages/go-kosu/rpc/service.go

@@ -3,6 +3,7 @@ package rpc
 import (
 	"context"
 	"go-kosu/abci"
+	"log"
 
 	"github.com/ethereum/go-ethereum/rpc"
 )
@@ -44,7 +45,10 @@ func (s *Service) Subscribe(ctx context.Context, query string) (*rpc.Subscriptio
 			case <-notifier.Closed():
 				return
 			case e := <-events:
-				notifier.Notify(rpcSub.ID, e)
+				err := notifier.Notify(rpcSub.ID, e)
+				if err != nil {
+					log.Printf("rpc: %+v", err)
+				}
 			}
 		}
 	}()

packages/go-kosu/tests/suite_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/suite"
 
 	"github.com/tendermint/tendermint/libs/db"
+	rpctypes "github.com/tendermint/tendermint/rpc/core/types"
 )
 
 type Suite struct {