+++ /dev/null
-package abcicli
-
-import (
- "bufio"
- "container/list"
- "errors"
- "fmt"
- "net"
- "reflect"
- "sync"
- "time"
-
- "github.com/tendermint/abci/types"
- cmn "github.com/tendermint/tmlibs/common"
-)
-
-const (
- OK = types.CodeType_OK
- LOG = ""
-)
-
-const reqQueueSize = 256 // TODO make configurable
-// const maxResponseSize = 1048576 // 1MB TODO make configurable
-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 socketClient struct {
- cmn.BaseService
-
- reqQueue chan *ReqRes
- flushTimer *cmn.ThrottleTimer
- mustConnect bool
-
- mtx sync.Mutex
- addr string
- conn net.Conn
- err error
- reqSent *list.List
- resCb func(*types.Request, *types.Response) // listens to all callbacks
-
-}
-
-func NewSocketClient(addr string, mustConnect bool) *socketClient {
- cli := &socketClient{
- reqQueue: make(chan *ReqRes, reqQueueSize),
- flushTimer: cmn.NewThrottleTimer("socketClient", flushThrottleMS),
- mustConnect: mustConnect,
-
- addr: addr,
- reqSent: list.New(),
- resCb: nil,
- }
- cli.BaseService = *cmn.NewBaseService(nil, "socketClient", cli)
- return cli
-}
-
-func (cli *socketClient) OnStart() error {
- if err := cli.BaseService.OnStart(); err != nil {
- return err
- }
-
- var err error
- var conn net.Conn
-RETRY_LOOP:
- for {
- conn, err = cmn.Connect(cli.addr)
- if err != nil {
- if cli.mustConnect {
- return err
- }
- cli.Logger.Error(fmt.Sprintf("abci.socketClient failed to connect to %v. Retrying...", cli.addr))
- time.Sleep(time.Second * 3)
- continue RETRY_LOOP
- }
- cli.conn = conn
-
- go cli.sendRequestsRoutine(conn)
- go cli.recvResponseRoutine(conn)
-
- return nil
- }
-}
-
-func (cli *socketClient) OnStop() {
- cli.BaseService.OnStop()
-
- cli.mtx.Lock()
- defer cli.mtx.Unlock()
- if cli.conn != nil {
- cli.conn.Close()
- }
-
- cli.flushQueue()
-}
-
-// Stop the client and set the error
-func (cli *socketClient) StopForError(err error) {
- if !cli.IsRunning() {
- return
- }
-
- cli.mtx.Lock()
- if cli.err == nil {
- cli.err = err
- }
- cli.mtx.Unlock()
-
- cli.Logger.Error(fmt.Sprintf("Stopping abci.socketClient for error: %v", err.Error()))
- cli.Stop()
-}
-
-func (cli *socketClient) Error() error {
- cli.mtx.Lock()
- defer cli.mtx.Unlock()
- return cli.err
-}
-
-// Set listener for all responses
-// NOTE: callback may get internally generated flush responses.
-func (cli *socketClient) SetResponseCallback(resCb Callback) {
- cli.mtx.Lock()
- defer cli.mtx.Unlock()
- cli.resCb = resCb
-}
-
-//----------------------------------------
-
-func (cli *socketClient) sendRequestsRoutine(conn net.Conn) {
-
- w := bufio.NewWriter(conn)
- for {
- select {
- case <-cli.flushTimer.Ch:
- select {
- case cli.reqQueue <- NewReqRes(types.ToRequestFlush()):
- default:
- // Probably will fill the buffer, or retry later.
- }
- case <-cli.BaseService.Quit:
- return
- case reqres := <-cli.reqQueue:
- cli.willSendReq(reqres)
- err := types.WriteMessage(reqres.Request, w)
- if err != nil {
- cli.StopForError(fmt.Errorf("Error writing msg: %v", err))
- return
- }
- // cli.Logger.Debug("Sent request", "requestType", reflect.TypeOf(reqres.Request), "request", reqres.Request)
- if _, ok := reqres.Request.Value.(*types.Request_Flush); ok {
- err = w.Flush()
- if err != nil {
- cli.StopForError(fmt.Errorf("Error flushing writer: %v", err))
- return
- }
- }
- }
- }
-}
-
-func (cli *socketClient) recvResponseRoutine(conn net.Conn) {
-
- r := bufio.NewReader(conn) // Buffer reads
- for {
- var res = &types.Response{}
- err := types.ReadMessage(r, res)
- if err != nil {
- cli.StopForError(err)
- return
- }
- switch r := res.Value.(type) {
- case *types.Response_Exception:
- // XXX After setting cli.err, release waiters (e.g. reqres.Done())
- cli.StopForError(errors.New(r.Exception.Error))
- return
- default:
- // cli.Logger.Debug("Received response", "responseType", reflect.TypeOf(res), "response", res)
- err := cli.didRecvResponse(res)
- if err != nil {
- cli.StopForError(err)
- return
- }
- }
- }
-}
-
-func (cli *socketClient) willSendReq(reqres *ReqRes) {
- cli.mtx.Lock()
- defer cli.mtx.Unlock()
- cli.reqSent.PushBack(reqres)
-}
-
-func (cli *socketClient) didRecvResponse(res *types.Response) error {
- cli.mtx.Lock()
- defer cli.mtx.Unlock()
-
- // Get the first ReqRes
- next := cli.reqSent.Front()
- if next == nil {
- return fmt.Errorf("Unexpected result type %v when nothing expected", reflect.TypeOf(res.Value))
- }
- reqres := next.Value.(*ReqRes)
- if !resMatchesReq(reqres.Request, res) {
- return fmt.Errorf("Unexpected result type %v when response to %v expected",
- reflect.TypeOf(res.Value), reflect.TypeOf(reqres.Request.Value))
- }
-
- reqres.Response = res // Set response
- reqres.Done() // Release waiters
- cli.reqSent.Remove(next) // Pop first item from linked list
-
- // Notify reqRes listener if set
- if cb := reqres.GetCallback(); cb != nil {
- cb(res)
- }
-
- // Notify client listener if set
- if cli.resCb != nil {
- cli.resCb(reqres.Request, res)
- }
-
- return nil
-}
-
-//----------------------------------------
-
-func (cli *socketClient) EchoAsync(msg string) *ReqRes {
- return cli.queueRequest(types.ToRequestEcho(msg))
-}
-
-func (cli *socketClient) FlushAsync() *ReqRes {
- return cli.queueRequest(types.ToRequestFlush())
-}
-
-func (cli *socketClient) InfoAsync(req types.RequestInfo) *ReqRes {
- return cli.queueRequest(types.ToRequestInfo(req))
-}
-
-func (cli *socketClient) SetOptionAsync(key string, value string) *ReqRes {
- return cli.queueRequest(types.ToRequestSetOption(key, value))
-}
-
-func (cli *socketClient) DeliverTxAsync(tx []byte) *ReqRes {
- return cli.queueRequest(types.ToRequestDeliverTx(tx))
-}
-
-func (cli *socketClient) CheckTxAsync(tx []byte) *ReqRes {
- return cli.queueRequest(types.ToRequestCheckTx(tx))
-}
-
-func (cli *socketClient) QueryAsync(reqQuery types.RequestQuery) *ReqRes {
- return cli.queueRequest(types.ToRequestQuery(reqQuery))
-}
-
-func (cli *socketClient) CommitAsync() *ReqRes {
- return cli.queueRequest(types.ToRequestCommit())
-}
-
-func (cli *socketClient) InitChainAsync(params types.RequestInitChain) *ReqRes {
- return cli.queueRequest(types.ToRequestInitChain(params))
-}
-
-func (cli *socketClient) BeginBlockAsync(params types.RequestBeginBlock) *ReqRes {
- return cli.queueRequest(types.ToRequestBeginBlock(params))
-}
-
-func (cli *socketClient) EndBlockAsync(height uint64) *ReqRes {
- return cli.queueRequest(types.ToRequestEndBlock(height))
-}
-
-//----------------------------------------
-
-func (cli *socketClient) EchoSync(msg string) (res types.Result) {
- reqres := cli.queueRequest(types.ToRequestEcho(msg))
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return types.ErrInternalError.SetLog(err.Error())
- }
- resp := reqres.Response.GetEcho()
- return types.Result{Code: OK, Data: []byte(resp.Message)}
-}
-
-func (cli *socketClient) FlushSync() error {
- reqRes := cli.queueRequest(types.ToRequestFlush())
- if err := cli.Error(); err != nil {
- return types.ErrInternalError.SetLog(err.Error())
- }
- reqRes.Wait() // NOTE: if we don't flush the queue, its possible to get stuck here
- return cli.Error()
-}
-
-func (cli *socketClient) InfoSync(req types.RequestInfo) (resInfo types.ResponseInfo, err error) {
- reqres := cli.queueRequest(types.ToRequestInfo(req))
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return resInfo, err
- }
- if resInfo_ := reqres.Response.GetInfo(); resInfo_ != nil {
- return *resInfo_, nil
- }
- return resInfo, nil
-}
-
-func (cli *socketClient) SetOptionSync(key string, value string) (res types.Result) {
- reqres := cli.queueRequest(types.ToRequestSetOption(key, value))
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return types.ErrInternalError.SetLog(err.Error())
- }
- resp := reqres.Response.GetSetOption()
- return types.Result{Code: OK, Data: nil, Log: resp.Log}
-}
-
-func (cli *socketClient) DeliverTxSync(tx []byte) (res types.Result) {
- reqres := cli.queueRequest(types.ToRequestDeliverTx(tx))
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return types.ErrInternalError.SetLog(err.Error())
- }
- resp := reqres.Response.GetDeliverTx()
- return types.Result{Code: resp.Code, Data: resp.Data, Log: resp.Log}
-}
-
-func (cli *socketClient) CheckTxSync(tx []byte) (res types.Result) {
- reqres := cli.queueRequest(types.ToRequestCheckTx(tx))
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return types.ErrInternalError.SetLog(err.Error())
- }
- resp := reqres.Response.GetCheckTx()
- return types.Result{Code: resp.Code, Data: resp.Data, Log: resp.Log}
-}
-
-func (cli *socketClient) QuerySync(reqQuery types.RequestQuery) (resQuery types.ResponseQuery, err error) {
- reqres := cli.queueRequest(types.ToRequestQuery(reqQuery))
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return resQuery, err
- }
- if resQuery_ := reqres.Response.GetQuery(); resQuery_ != nil {
- return *resQuery_, nil
- }
- return resQuery, nil
-}
-
-func (cli *socketClient) CommitSync() (res types.Result) {
- reqres := cli.queueRequest(types.ToRequestCommit())
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return types.ErrInternalError.SetLog(err.Error())
- }
- resp := reqres.Response.GetCommit()
- return types.Result{Code: resp.Code, Data: resp.Data, Log: resp.Log}
-}
-
-func (cli *socketClient) InitChainSync(params types.RequestInitChain) (err error) {
- cli.queueRequest(types.ToRequestInitChain(params))
- cli.FlushSync()
- return cli.Error()
-}
-
-func (cli *socketClient) BeginBlockSync(params types.RequestBeginBlock) (err error) {
- cli.queueRequest(types.ToRequestBeginBlock(params))
- cli.FlushSync()
- return cli.Error()
-}
-
-func (cli *socketClient) EndBlockSync(height uint64) (resEndBlock types.ResponseEndBlock, err error) {
- reqres := cli.queueRequest(types.ToRequestEndBlock(height))
- cli.FlushSync()
- if err := cli.Error(); err != nil {
- return resEndBlock, err
- }
- if blk := reqres.Response.GetEndBlock(); blk != nil {
- return *blk, nil
- }
- return resEndBlock, nil
-}
-
-//----------------------------------------
-
-func (cli *socketClient) queueRequest(req *types.Request) *ReqRes {
- reqres := NewReqRes(req)
-
- // TODO: set cli.err if reqQueue times out
- cli.reqQueue <- reqres
-
- // Maybe auto-flush, or unset auto-flush
- switch req.Value.(type) {
- case *types.Request_Flush:
- cli.flushTimer.Unset()
- default:
- cli.flushTimer.Set()
- }
-
- return reqres
-}
-
-func (cli *socketClient) flushQueue() {
-LOOP:
- for {
- select {
- case reqres := <-cli.reqQueue:
- reqres.Done()
- default:
- break LOOP
- }
- }
-}
-
-//----------------------------------------
-
-func resMatchesReq(req *types.Request, res *types.Response) (ok bool) {
- switch req.Value.(type) {
- case *types.Request_Echo:
- _, ok = res.Value.(*types.Response_Echo)
- case *types.Request_Flush:
- _, ok = res.Value.(*types.Response_Flush)
- case *types.Request_Info:
- _, ok = res.Value.(*types.Response_Info)
- case *types.Request_SetOption:
- _, ok = res.Value.(*types.Response_SetOption)
- case *types.Request_DeliverTx:
- _, ok = res.Value.(*types.Response_DeliverTx)
- case *types.Request_CheckTx:
- _, ok = res.Value.(*types.Response_CheckTx)
- case *types.Request_Commit:
- _, ok = res.Value.(*types.Response_Commit)
- case *types.Request_Query:
- _, ok = res.Value.(*types.Response_Query)
- case *types.Request_InitChain:
- _, ok = res.Value.(*types.Response_InitChain)
- case *types.Request_BeginBlock:
- _, ok = res.Value.(*types.Response_BeginBlock)
- case *types.Request_EndBlock:
- _, ok = res.Value.(*types.Response_EndBlock)
- }
- return ok
-}
OSDN Git Service
