go-kosu: subscriptio and call PoC

packages/go-kosu/rpc/client.go

@@ -0,0 +1,47 @@
+package rpc
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+type Client struct {
+	rpc *rpc.Client
+}
+
+func DialInProc(srv *rpc.Server) *Client {
+	return &Client{
+		rpc: rpc.DialInProc(srv),
+	}
+}
+
+func (c *Client) Subscribe(ctx context.Context, fn func(interface{}), query string) error {
+	ch := make(chan interface{})
+	args := []interface{}{"subscribe", query}
+	sub, err := c.rpc.Subscribe(ctx, "kosu", ch, args...)
+	if err != nil {
+		return err
+	}
+
+	go func() {
+		defer close(ch)
+		defer sub.Unsubscribe()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-sub.Err():
+				return
+			case i := <-ch:
+				fn(i)
+			}
+		}
+	}()
+	return nil
+}
+
+func (c *Client) Call(result interface{}, method string, args ...interface{}) error {
+	return c.rpc.Call(result, method, args...)
+}

packages/go-kosu/rpc/rpc.go

@@ -0,0 +1,13 @@
+package rpc
+
+import (
+	"go-kosu/abci"
+
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+func NewServer(abci *abci.Client) *rpc.Server {
+	srv := rpc.NewServer()
+	srv.RegisterName("kosu", &Service{abci: abci})
+	return srv
+}

packages/go-kosu/rpc/rpc_test.go

@@ -0,0 +1,65 @@
+package rpc
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"testing"
+	"time"
+
+	"go-kosu/abci"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tendermint/tendermint/libs/db"
+	"github.com/tendermint/tendermint/libs/log"
+)
+
+func TestRPCCall(t *testing.T) {
+	server := NewServer(nil)
+	client := DialInProc(server)
+	require.NoError(t, client.Call(nil, "kosu_foo"))
+}
+
+func TestRPCSubscription(t *testing.T) {
+	_, closer := startServer(t, db.NewMemDB())
+	defer closer()
+	abciClient := abci.NewHTTPClient("http://localhost:26657", nil)
+
+	server := NewServer(abciClient)
+	client := DialInProc(server)
+
+	fn := func(i interface{}) {
+		fmt.Printf("i = %+v\n", i)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+
+	err := client.Subscribe(ctx, fn, "tm.event = 'NewBlock'")
+	require.NoError(t, err)
+
+	time.Sleep(3 * time.Second)
+	cancel()
+}
+
+// TODO use tests/support.go version of startServer
+func startServer(t *testing.T, db db.DB) (*abci.App, func()) {
+	// Create a temp dir and initialize tendermint there
+	dir, err := ioutil.TempDir("/tmp", "/go-kosu-go-tests_")
+	require.NoError(t, err)
+
+	err = abci.InitTendermintWithLogger(dir, log.NewNopLogger())
+	require.NoError(t, err)
+
+	// Initialize the server
+	app := abci.NewApp(db, dir)
+	app.Config.LogFormat = "none"
+	app.Config.LogLevel = "app:error"
+	srv, err := abci.StartInProcessServer(app)
+	require.NoError(t, err)
+
+	// nolint
+	return app, func() {
+		srv.Stop()
+		os.RemoveAll(dir)
+	}
+}

packages/go-kosu/rpc/service.go

@@ -0,0 +1,57 @@
+package rpc
+
+import (
+	"context"
+	"go-kosu/abci"
+
+	"github.com/ethereum/go-ethereum/rpc"
+)
+
+// Service is a RPC service
+type Service struct {
+	abci *abci.Client
+}
+
+// NewService returns a new service given a abci client
+func NewService(abci *abci.Client) *Service {
+	return &Service{
+		abci: abci,
+	}
+}
+
+// Subscribe subscribes to the ABCI events
+// To tell which events you want, you need to provide a query.
+// More information about query can be found here: https://tendermint.com/rpc/#subscribe
+func (s *Service) Subscribe(ctx context.Context, query string) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return nil, rpc.ErrNotificationsUnsupported
+	}
+
+	events, closer, err := s.abci.Subscribe(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	rpcSub := notifier.CreateSubscription()
+	go func() {
+		defer closer()
+
+		for {
+			select {
+			case <-rpcSub.Err():
+				return
+			case <-notifier.Closed():
+				return
+			case e := <-events:
+				notifier.Notify(rpcSub.ID, e)
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}
+
+func (s *Service) Foo() error {
+	return nil
+}