Use TMSPClient

proxy/app_conn.go

@@ -1,13 +1,11 @@
 package proxy
 
 import (
-	tmsp "github.com/tendermint/tmsp/types"
+	tmspcli "github.com/tendermint/tmsp/client/golang"
 )
 
-type Callback func(tmsp.Request, tmsp.Response)
-
 type AppConn interface {
-	SetResponseCallback(Callback)
+	SetResponseCallback(tmspcli.Callback)
 	Error() error
 
 	EchoAsync(msg string)

proxy/local_app_conn.go

@@ -1,6 +1,7 @@
 package proxy
 
 import (
+	tmspcli "github.com/tendermint/tmsp/client/golang"
 	tmsp "github.com/tendermint/tmsp/types"
 	"sync"
 )
@@ -8,7 +9,7 @@ import (
 type localAppConn struct {
 	mtx *sync.Mutex
 	tmsp.Application
-	Callback
+	tmspcli.Callback
 }
 
 func NewLocalAppConn(mtx *sync.Mutex, app tmsp.Application) *localAppConn {
@@ -18,7 +19,7 @@ func NewLocalAppConn(mtx *sync.Mutex, app tmsp.Application) *localAppConn {
 	}
 }
 
-func (app *localAppConn) SetResponseCallback(cb Callback) {
+func (app *localAppConn) SetResponseCallback(cb tmspcli.Callback) {
 	app.mtx.Lock()
 	defer app.mtx.Unlock()
 	app.Callback = cb

proxy/remote_app_conn.go

@@ -1,307 +1,21 @@
 package proxy
 
 import (
-	"bufio"
-	"container/list"
-	"errors"
-	"fmt"
 	"net"
-	"reflect"
-	"sync"
 
-	. "github.com/tendermint/go-common"
-	"github.com/tendermint/go-wire"
-	tmsp "github.com/tendermint/tmsp/types"
+	tmspcli "github.com/tendermint/tmsp/client/golang"
 )
 
-const maxResponseSize = 1048576 // 1MB
-const flushThrottleMS = 20      // Don't wait longer than...
-
 // This is goroutine-safe, but users should beware that
 // the application in general is not meant to be interfaced
 // with concurrent callers.
 type remoteAppConn struct {
-	QuitService
-	sync.Mutex // [EB]: is this even used?
-
-	reqQueue   chan *reqRes
-	flushTimer *ThrottleTimer
-
-	mtx       sync.Mutex
-	conn      net.Conn
-	bufWriter *bufio.Writer
-	err       error
-	reqSent   *list.List
-	resCb     func(tmsp.Request, tmsp.Response)
+	*tmspcli.TMSPClient
 }
 
 func NewRemoteAppConn(conn net.Conn, bufferSize int) *remoteAppConn {
 	app := &remoteAppConn{
-		reqQueue:   make(chan *reqRes, bufferSize),
-		flushTimer: NewThrottleTimer("remoteAppConn", flushThrottleMS),
-
-		conn:      conn,
-		bufWriter: bufio.NewWriter(conn),
-		reqSent:   list.New(),
-		resCb:     nil,
+		TMSPClient: tmspcli.NewTMSPClient(conn, bufferSize),
 	}
-	app.QuitService = *NewQuitService(nil, "remoteAppConn", app)
 	return app
 }
-
-func (app *remoteAppConn) OnStart() error {
-	app.QuitService.OnStart()
-	go app.sendRequestsRoutine()
-	go app.recvResponseRoutine()
-	return nil
-}
-
-func (app *remoteAppConn) OnStop() {
-	app.QuitService.OnStop()
-	app.conn.Close()
-}
-
-// NOTE: callback may get internally generated flush responses.
-func (app *remoteAppConn) SetResponseCallback(resCb Callback) {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
-	app.resCb = resCb
-}
-
-func (app *remoteAppConn) StopForError(err error) {
-	app.mtx.Lock()
-	log.Error("Stopping remoteAppConn for error.", "error", err)
-	if app.err == nil {
-		app.err = err
-	}
-	app.mtx.Unlock()
-	app.Stop()
-}
-
-func (app *remoteAppConn) Error() error {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
-	return app.err
-}
-
-//----------------------------------------
-
-func (app *remoteAppConn) sendRequestsRoutine() {
-	for {
-		var n int
-		var err error
-		select {
-		case <-app.flushTimer.Ch:
-			select {
-			case app.reqQueue <- newReqRes(tmsp.RequestFlush{}):
-			default:
-				// Probably will fill the buffer, or retry later.
-			}
-		case <-app.QuitService.Quit:
-			return
-		case reqres := <-app.reqQueue:
-			app.willSendReq(reqres)
-			wire.WriteBinaryLengthPrefixed(struct{ tmsp.Request }{reqres.Request}, app.bufWriter, &n, &err) // Length prefix
-			if err != nil {
-				app.StopForError(err)
-				return
-			}
-			log.Debug("Sent request", "requestType", reflect.TypeOf(reqres.Request), "request", reqres.Request)
-			if _, ok := reqres.Request.(tmsp.RequestFlush); ok {
-				err = app.bufWriter.Flush()
-				if err != nil {
-					app.StopForError(err)
-					return
-				}
-			}
-		}
-	}
-}
-
-func (app *remoteAppConn) recvResponseRoutine() {
-	r := bufio.NewReader(app.conn) // Buffer reads
-	for {
-		var res tmsp.Response
-		var n int
-		var err error
-		wire.ReadBinaryPtrLengthPrefixed(&res, r, maxResponseSize, &n, &err)
-		if err != nil {
-			app.StopForError(err)
-			return
-		}
-		switch res := res.(type) {
-		case tmsp.ResponseException:
-			app.StopForError(errors.New(res.Error))
-		default:
-			log.Debug("Received response", "responseType", reflect.TypeOf(res), "response", res)
-			err := app.didRecvResponse(res)
-			if err != nil {
-				app.StopForError(err)
-			}
-		}
-	}
-}
-
-func (app *remoteAppConn) willSendReq(reqres *reqRes) {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
-	app.reqSent.PushBack(reqres)
-}
-
-func (app *remoteAppConn) didRecvResponse(res tmsp.Response) error {
-	app.mtx.Lock()
-	defer app.mtx.Unlock()
-
-	// Special logic for events which have no corresponding requests.
-	if _, ok := res.(tmsp.ResponseEvent); ok && app.resCb != nil {
-		app.resCb(nil, res)
-		return nil
-	}
-
-	// Get the first reqRes
-	next := app.reqSent.Front()
-	if next == nil {
-		return fmt.Errorf("Unexpected result type %v when nothing expected", reflect.TypeOf(res))
-	}
-	reqres := next.Value.(*reqRes)
-	if !resMatchesReq(reqres.Request, res) {
-		return fmt.Errorf("Unexpected result type %v when response to %v expected",
-			reflect.TypeOf(res), reflect.TypeOf(reqres.Request))
-	}
-
-	reqres.Response = res    // Set response
-	reqres.Done()            // Release waiters
-	app.reqSent.Remove(next) // Pop first item from linked list
-
-	// Callback if there is a listener
-	if app.resCb != nil {
-		app.resCb(reqres.Request, res)
-	}
-
-	return nil
-}
-
-//----------------------------------------
-
-func (app *remoteAppConn) EchoAsync(msg string) {
-	app.queueRequest(tmsp.RequestEcho{msg})
-}
-
-func (app *remoteAppConn) FlushAsync() {
-	app.queueRequest(tmsp.RequestFlush{})
-}
-
-func (app *remoteAppConn) SetOptionAsync(key string, value string) {
-	app.queueRequest(tmsp.RequestSetOption{key, value})
-}
-
-func (app *remoteAppConn) AppendTxAsync(tx []byte) {
-	app.queueRequest(tmsp.RequestAppendTx{tx})
-}
-
-func (app *remoteAppConn) CheckTxAsync(tx []byte) {
-	app.queueRequest(tmsp.RequestCheckTx{tx})
-}
-
-func (app *remoteAppConn) GetHashAsync() {
-	app.queueRequest(tmsp.RequestGetHash{})
-}
-
-func (app *remoteAppConn) AddListenerAsync(key string) {
-	app.queueRequest(tmsp.RequestAddListener{key})
-}
-
-func (app *remoteAppConn) RemListenerAsync(key string) {
-	app.queueRequest(tmsp.RequestRemListener{key})
-}
-
-//----------------------------------------
-
-func (app *remoteAppConn) InfoSync() (info []string, err error) {
-	reqres := app.queueRequest(tmsp.RequestInfo{})
-	app.FlushSync()
-	if app.err != nil {
-		return nil, app.err
-	}
-	return reqres.Response.(tmsp.ResponseInfo).Data, nil
-}
-
-func (app *remoteAppConn) FlushSync() error {
-	app.queueRequest(tmsp.RequestFlush{}).Wait()
-	return app.err
-}
-
-func (app *remoteAppConn) GetHashSync() (hash []byte, err error) {
-	reqres := app.queueRequest(tmsp.RequestGetHash{})
-	app.FlushSync()
-	if app.err != nil {
-		return nil, app.err
-	}
-	return reqres.Response.(tmsp.ResponseGetHash).Hash, nil
-}
-
-//----------------------------------------
-
-func (app *remoteAppConn) queueRequest(req tmsp.Request) *reqRes {
-	reqres := newReqRes(req)
-	// TODO: set app.err if reqQueue times out
-	app.reqQueue <- reqres
-
-	// Maybe auto-flush, or unset auto-flush
-	switch req.(type) {
-	case tmsp.RequestFlush:
-		app.flushTimer.Unset()
-	default:
-		app.flushTimer.Set()
-	}
-
-	return reqres
-}
-
-//----------------------------------------
-
-func resMatchesReq(req tmsp.Request, res tmsp.Response) (ok bool) {
-	switch req.(type) {
-	case tmsp.RequestEcho:
-		_, ok = res.(tmsp.ResponseEcho)
-	case tmsp.RequestFlush:
-		_, ok = res.(tmsp.ResponseFlush)
-	case tmsp.RequestInfo:
-		_, ok = res.(tmsp.ResponseInfo)
-	case tmsp.RequestSetOption:
-		_, ok = res.(tmsp.ResponseSetOption)
-	case tmsp.RequestAppendTx:
-		_, ok = res.(tmsp.ResponseAppendTx)
-	case tmsp.RequestCheckTx:
-		_, ok = res.(tmsp.ResponseCheckTx)
-	case tmsp.RequestGetHash:
-		_, ok = res.(tmsp.ResponseGetHash)
-	case tmsp.RequestAddListener:
-		_, ok = res.(tmsp.ResponseAddListener)
-	case tmsp.RequestRemListener:
-		_, ok = res.(tmsp.ResponseRemListener)
-	default:
-		return false
-	}
-	return
-}
-
-type reqRes struct {
-	tmsp.Request
-	*sync.WaitGroup
-	tmsp.Response // Not set atomically, so be sure to use WaitGroup.
-}
-
-func newReqRes(req tmsp.Request) *reqRes {
-	return &reqRes{
-		Request:   req,
-		WaitGroup: waitGroup1(),
-		Response:  nil,
-	}
-}
-
-func waitGroup1() (wg *sync.WaitGroup) {
-	wg = &sync.WaitGroup{}
-	wg.Add(1)
-	return
-}