From 26d64208d18991861a338ccc540021a93be3efb8 Mon Sep 17 00:00:00 2001 From: Jae Kwon Date: Fri, 22 Jan 2016 15:48:13 -0800 Subject: [PATCH] Use TMSPClient --- proxy/app_conn.go | 6 +- proxy/local_app_conn.go | 5 +- proxy/remote_app_conn.go | 292 +-------------------------------------- 3 files changed, 8 insertions(+), 295 deletions(-) diff --git a/proxy/app_conn.go b/proxy/app_conn.go index 3b98eb608..e4f37e300 100644 --- a/proxy/app_conn.go +++ b/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) diff --git a/proxy/local_app_conn.go b/proxy/local_app_conn.go index 680bb764f..355cef787 100644 --- a/proxy/local_app_conn.go +++ b/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 diff --git a/proxy/remote_app_conn.go b/proxy/remote_app_conn.go index a1dff9a84..c773d2bba 100644 --- a/proxy/remote_app_conn.go +++ b/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 -}