You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

307 lines
7.1 KiB

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"
)
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)
}
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,
}
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
}