+++ /dev/null
-// Copyright (c) 2014-2017 The btcsuite developers
-// Use of this source code is governed by an ISC
-// license that can be found in the LICENSE file.
-
-package rpcclient
-
-import (
- "bytes"
- "container/list"
- "crypto/tls"
- "crypto/x509"
- "encoding/base64"
- "encoding/json"
- "errors"
- "fmt"
- "io"
- "io/ioutil"
- "math"
- "net"
- "net/http"
- "net/url"
- "sync"
- "sync/atomic"
- "time"
-
- "github.com/btcsuite/btcd/btcjson"
- "github.com/btcsuite/go-socks/socks"
- "github.com/btcsuite/websocket"
-)
-
-var (
- // ErrInvalidAuth is an error to describe the condition where the client
- // is either unable to authenticate or the specified endpoint is
- // incorrect.
- ErrInvalidAuth = errors.New("authentication failure")
-
- // ErrInvalidEndpoint is an error to describe the condition where the
- // websocket handshake failed with the specified endpoint.
- ErrInvalidEndpoint = errors.New("the endpoint either does not support " +
- "websockets or does not exist")
-
- // ErrClientNotConnected is an error to describe the condition where a
- // websocket client has been created, but the connection was never
- // established. This condition differs from ErrClientDisconnect, which
- // represents an established connection that was lost.
- ErrClientNotConnected = errors.New("the client was never connected")
-
- // ErrClientDisconnect is an error to describe the condition where the
- // client has been disconnected from the RPC server. When the
- // DisableAutoReconnect option is not set, any outstanding futures
- // when a client disconnect occurs will return this error as will
- // any new requests.
- ErrClientDisconnect = errors.New("the client has been disconnected")
-
- // ErrClientShutdown is an error to describe the condition where the
- // client is either already shutdown, or in the process of shutting
- // down. Any outstanding futures when a client shutdown occurs will
- // return this error as will any new requests.
- ErrClientShutdown = errors.New("the client has been shutdown")
-
- // ErrNotWebsocketClient is an error to describe the condition of
- // calling a Client method intended for a websocket client when the
- // client has been configured to run in HTTP POST mode instead.
- ErrNotWebsocketClient = errors.New("client is not configured for " +
- "websockets")
-
- // ErrClientAlreadyConnected is an error to describe the condition where
- // a new client connection cannot be established due to a websocket
- // client having already connected to the RPC server.
- ErrClientAlreadyConnected = errors.New("websocket client has already " +
- "connected")
-)
-
-const (
- // sendBufferSize is the number of elements the websocket send channel
- // can queue before blocking.
- sendBufferSize = 50
-
- // sendPostBufferSize is the number of elements the HTTP POST send
- // channel can queue before blocking.
- sendPostBufferSize = 100
-
- // connectionRetryInterval is the amount of time to wait in between
- // retries when automatically reconnecting to an RPC server.
- connectionRetryInterval = time.Second * 5
-)
-
-// sendPostDetails houses an HTTP POST request to send to an RPC server as well
-// as the original JSON-RPC command and a channel to reply on when the server
-// responds with the result.
-type sendPostDetails struct {
- httpRequest *http.Request
- jsonRequest *jsonRequest
-}
-
-// jsonRequest holds information about a json request that is used to properly
-// detect, interpret, and deliver a reply to it.
-type jsonRequest struct {
- id uint64
- method string
- cmd interface{}
- marshalledJSON []byte
- responseChan chan *response
-}
-
-// Client represents a Bitcoin RPC client which allows easy access to the
-// various RPC methods available on a Bitcoin RPC server. Each of the wrapper
-// functions handle the details of converting the passed and return types to and
-// from the underlying JSON types which are required for the JSON-RPC
-// invocations
-//
-// The client provides each RPC in both synchronous (blocking) and asynchronous
-// (non-blocking) forms. The asynchronous forms are based on the concept of
-// futures where they return an instance of a type that promises to deliver the
-// result of the invocation at some future time. Invoking the Receive method on
-// the returned future will block until the result is available if it's not
-// already.
-type Client struct {
- id uint64 // atomic, so must stay 64-bit aligned
-
- // config holds the connection configuration assoiated with this client.
- config *ConnConfig
-
- // wsConn is the underlying websocket connection when not in HTTP POST
- // mode.
- wsConn *websocket.Conn
-
- // httpClient is the underlying HTTP client to use when running in HTTP
- // POST mode.
- httpClient *http.Client
-
- // mtx is a mutex to protect access to connection related fields.
- mtx sync.Mutex
-
- // disconnected indicated whether or not the server is disconnected.
- disconnected bool
-
- // retryCount holds the number of times the client has tried to
- // reconnect to the RPC server.
- retryCount int64
-
- // Track command and their response channels by ID.
- requestLock sync.Mutex
- requestMap map[uint64]*list.Element
- requestList *list.List
-
- // Notifications.
- ntfnHandlers *NotificationHandlers
- ntfnStateLock sync.Mutex
- ntfnState *notificationState
-
- // Networking infrastructure.
- sendChan chan []byte
- sendPostChan chan *sendPostDetails
- connEstablished chan struct{}
- disconnect chan struct{}
- shutdown chan struct{}
- wg sync.WaitGroup
-}
-
-// NextID returns the next id to be used when sending a JSON-RPC message. This
-// ID allows responses to be associated with particular requests per the
-// JSON-RPC specification. Typically the consumer of the client does not need
-// to call this function, however, if a custom request is being created and used
-// this function should be used to ensure the ID is unique amongst all requests
-// being made.
-func (c *Client) NextID() uint64 {
- return atomic.AddUint64(&c.id, 1)
-}
-
-// addRequest associates the passed jsonRequest with its id. This allows the
-// response from the remote server to be unmarshalled to the appropriate type
-// and sent to the specified channel when it is received.
-//
-// If the client has already begun shutting down, ErrClientShutdown is returned
-// and the request is not added.
-//
-// This function is safe for concurrent access.
-func (c *Client) addRequest(jReq *jsonRequest) error {
- c.requestLock.Lock()
- defer c.requestLock.Unlock()
-
- // A non-blocking read of the shutdown channel with the request lock
- // held avoids adding the request to the client's internal data
- // structures if the client is in the process of shutting down (and
- // has not yet grabbed the request lock), or has finished shutdown
- // already (responding to each outstanding request with
- // ErrClientShutdown).
- select {
- case <-c.shutdown:
- return ErrClientShutdown
- default:
- }
-
- element := c.requestList.PushBack(jReq)
- c.requestMap[jReq.id] = element
- return nil
-}
-
-// removeRequest returns and removes the jsonRequest which contains the response
-// channel and original method associated with the passed id or nil if there is
-// no association.
-//
-// This function is safe for concurrent access.
-func (c *Client) removeRequest(id uint64) *jsonRequest {
- c.requestLock.Lock()
- defer c.requestLock.Unlock()
-
- element := c.requestMap[id]
- if element != nil {
- delete(c.requestMap, id)
- request := c.requestList.Remove(element).(*jsonRequest)
- return request
- }
-
- return nil
-}
-
-// removeAllRequests removes all the jsonRequests which contain the response
-// channels for outstanding requests.
-//
-// This function MUST be called with the request lock held.
-func (c *Client) removeAllRequests() {
- c.requestMap = make(map[uint64]*list.Element)
- c.requestList.Init()
-}
-
-// trackRegisteredNtfns examines the passed command to see if it is one of
-// the notification commands and updates the notification state that is used
-// to automatically re-establish registered notifications on reconnects.
-func (c *Client) trackRegisteredNtfns(cmd interface{}) {
- // Nothing to do if the caller is not interested in notifications.
- if c.ntfnHandlers == nil {
- return
- }
-
- c.ntfnStateLock.Lock()
- defer c.ntfnStateLock.Unlock()
-
- switch bcmd := cmd.(type) {
- case *btcjson.NotifyBlocksCmd:
- c.ntfnState.notifyBlocks = true
-
- case *btcjson.NotifyNewTransactionsCmd:
- if bcmd.Verbose != nil && *bcmd.Verbose {
- c.ntfnState.notifyNewTxVerbose = true
- } else {
- c.ntfnState.notifyNewTx = true
-
- }
-
- case *btcjson.NotifySpentCmd:
- for _, op := range bcmd.OutPoints {
- c.ntfnState.notifySpent[op] = struct{}{}
- }
-
- case *btcjson.NotifyReceivedCmd:
- for _, addr := range bcmd.Addresses {
- c.ntfnState.notifyReceived[addr] = struct{}{}
- }
- }
-}
-
-type (
- // inMessage is the first type that an incoming message is unmarshaled
- // into. It supports both requests (for notification support) and
- // responses. The partially-unmarshaled message is a notification if
- // the embedded ID (from the response) is nil. Otherwise, it is a
- // response.
- inMessage struct {
- ID *float64 `json:"id"`
- *rawNotification
- *rawResponse
- }
-
- // rawNotification is a partially-unmarshaled JSON-RPC notification.
- rawNotification struct {
- Method string `json:"method"`
- Params []json.RawMessage `json:"params"`
- }
-
- // rawResponse is a partially-unmarshaled JSON-RPC response. For this
- // to be valid (according to JSON-RPC 1.0 spec), ID may not be nil.
- rawResponse struct {
- Result json.RawMessage `json:"result"`
- Error *btcjson.RPCError `json:"error"`
- }
-)
-
-// response is the raw bytes of a JSON-RPC result, or the error if the response
-// error object was non-null.
-type response struct {
- result []byte
- err error
-}
-
-// result checks whether the unmarshaled response contains a non-nil error,
-// returning an unmarshaled btcjson.RPCError (or an unmarshaling error) if so.
-// If the response is not an error, the raw bytes of the request are
-// returned for further unmashaling into specific result types.
-func (r rawResponse) result() (result []byte, err error) {
- if r.Error != nil {
- return nil, r.Error
- }
- return r.Result, nil
-}
-
-// handleMessage is the main handler for incoming notifications and responses.
-func (c *Client) handleMessage(msg []byte) {
- // Attempt to unmarshal the message as either a notification or
- // response.
- var in inMessage
- err := json.Unmarshal(msg, &in)
- if err != nil {
- log.Warnf("Remote server sent invalid message: %v", err)
- return
- }
-
- // JSON-RPC 1.0 notifications are requests with a null id.
- if in.ID == nil {
- ntfn := in.rawNotification
- if ntfn == nil {
- log.Warn("Malformed notification: missing " +
- "method and parameters")
- return
- }
- if ntfn.Method == "" {
- log.Warn("Malformed notification: missing method")
- return
- }
- // params are not optional: nil isn't valid (but len == 0 is)
- if ntfn.Params == nil {
- log.Warn("Malformed notification: missing params")
- return
- }
- // Deliver the notification.
- log.Tracef("Received notification [%s]", in.Method)
- c.handleNotification(in.rawNotification)
- return
- }
-
- // ensure that in.ID can be converted to an integer without loss of precision
- if *in.ID < 0 || *in.ID != math.Trunc(*in.ID) {
- log.Warn("Malformed response: invalid identifier")
- return
- }
-
- if in.rawResponse == nil {
- log.Warn("Malformed response: missing result and error")
- return
- }
-
- id := uint64(*in.ID)
- log.Tracef("Received response for id %d (result %s)", id, in.Result)
- request := c.removeRequest(id)
-
- // Nothing more to do if there is no request associated with this reply.
- if request == nil || request.responseChan == nil {
- log.Warnf("Received unexpected reply: %s (id %d)", in.Result,
- id)
- return
- }
-
- // Since the command was successful, examine it to see if it's a
- // notification, and if is, add it to the notification state so it
- // can automatically be re-established on reconnect.
- c.trackRegisteredNtfns(request.cmd)
-
- // Deliver the response.
- result, err := in.rawResponse.result()
- request.responseChan <- &response{result: result, err: err}
-}
-
-// shouldLogReadError returns whether or not the passed error, which is expected
-// to have come from reading from the websocket connection in wsInHandler,
-// should be logged.
-func (c *Client) shouldLogReadError(err error) bool {
- // No logging when the connetion is being forcibly disconnected.
- select {
- case <-c.shutdown:
- return false
- default:
- }
-
- // No logging when the connection has been disconnected.
- if err == io.EOF {
- return false
- }
- if opErr, ok := err.(*net.OpError); ok && !opErr.Temporary() {
- return false
- }
-
- return true
-}
-
-// wsInHandler handles all incoming messages for the websocket connection
-// associated with the client. It must be run as a goroutine.
-func (c *Client) wsInHandler() {
-out:
- for {
- // Break out of the loop once the shutdown channel has been
- // closed. Use a non-blocking select here so we fall through
- // otherwise.
- select {
- case <-c.shutdown:
- break out
- default:
- }
-
- _, msg, err := c.wsConn.ReadMessage()
- if err != nil {
- // Log the error if it's not due to disconnecting.
- if c.shouldLogReadError(err) {
- log.Errorf("Websocket receive error from "+
- "%s: %v", c.config.Host, err)
- }
- break out
- }
- c.handleMessage(msg)
- }
-
- // Ensure the connection is closed.
- c.Disconnect()
- c.wg.Done()
- log.Tracef("RPC client input handler done for %s", c.config.Host)
-}
-
-// disconnectChan returns a copy of the current disconnect channel. The channel
-// is read protected by the client mutex, and is safe to call while the channel
-// is being reassigned during a reconnect.
-func (c *Client) disconnectChan() <-chan struct{} {
- c.mtx.Lock()
- ch := c.disconnect
- c.mtx.Unlock()
- return ch
-}
-
-// wsOutHandler handles all outgoing messages for the websocket connection. It
-// uses a buffered channel to serialize output messages while allowing the
-// sender to continue running asynchronously. It must be run as a goroutine.
-func (c *Client) wsOutHandler() {
-out:
- for {
- // Send any messages ready for send until the client is
- // disconnected closed.
- select {
- case msg := <-c.sendChan:
- err := c.wsConn.WriteMessage(websocket.TextMessage, msg)
- if err != nil {
- c.Disconnect()
- break out
- }
-
- case <-c.disconnectChan():
- break out
- }
- }
-
- // Drain any channels before exiting so nothing is left waiting around
- // to send.
-cleanup:
- for {
- select {
- case <-c.sendChan:
- default:
- break cleanup
- }
- }
- c.wg.Done()
- log.Tracef("RPC client output handler done for %s", c.config.Host)
-}
-
-// sendMessage sends the passed JSON to the connected server using the
-// websocket connection. It is backed by a buffered channel, so it will not
-// block until the send channel is full.
-func (c *Client) sendMessage(marshalledJSON []byte) {
- // Don't send the message if disconnected.
- select {
- case c.sendChan <- marshalledJSON:
- case <-c.disconnectChan():
- return
- }
-}
-
-// reregisterNtfns creates and sends commands needed to re-establish the current
-// notification state associated with the client. It should only be called on
-// on reconnect by the resendRequests function.
-func (c *Client) reregisterNtfns() error {
- // Nothing to do if the caller is not interested in notifications.
- if c.ntfnHandlers == nil {
- return nil
- }
-
- // In order to avoid holding the lock on the notification state for the
- // entire time of the potentially long running RPCs issued below, make a
- // copy of it and work from that.
- //
- // Also, other commands will be running concurrently which could modify
- // the notification state (while not under the lock of course) which
- // also register it with the remote RPC server, so this prevents double
- // registrations.
- c.ntfnStateLock.Lock()
- stateCopy := c.ntfnState.Copy()
- c.ntfnStateLock.Unlock()
-
- // Reregister notifyblocks if needed.
- if stateCopy.notifyBlocks {
- log.Debugf("Reregistering [notifyblocks]")
- if err := c.NotifyBlocks(); err != nil {
- return err
- }
- }
-
- // Reregister notifynewtransactions if needed.
- if stateCopy.notifyNewTx || stateCopy.notifyNewTxVerbose {
- log.Debugf("Reregistering [notifynewtransactions] (verbose=%v)",
- stateCopy.notifyNewTxVerbose)
- err := c.NotifyNewTransactions(stateCopy.notifyNewTxVerbose)
- if err != nil {
- return err
- }
- }
-
- // Reregister the combination of all previously registered notifyspent
- // outpoints in one command if needed.
- nslen := len(stateCopy.notifySpent)
- if nslen > 0 {
- outpoints := make([]btcjson.OutPoint, 0, nslen)
- for op := range stateCopy.notifySpent {
- outpoints = append(outpoints, op)
- }
- log.Debugf("Reregistering [notifyspent] outpoints: %v", outpoints)
- if err := c.notifySpentInternal(outpoints).Receive(); err != nil {
- return err
- }
- }
-
- // Reregister the combination of all previously registered
- // notifyreceived addresses in one command if needed.
- nrlen := len(stateCopy.notifyReceived)
- if nrlen > 0 {
- addresses := make([]string, 0, nrlen)
- for addr := range stateCopy.notifyReceived {
- addresses = append(addresses, addr)
- }
- log.Debugf("Reregistering [notifyreceived] addresses: %v", addresses)
- if err := c.notifyReceivedInternal(addresses).Receive(); err != nil {
- return err
- }
- }
-
- return nil
-}
-
-// ignoreResends is a set of all methods for requests that are "long running"
-// are not be reissued by the client on reconnect.
-var ignoreResends = map[string]struct{}{
- "rescan": {},
-}
-
-// resendRequests resends any requests that had not completed when the client
-// disconnected. It is intended to be called once the client has reconnected as
-// a separate goroutine.
-func (c *Client) resendRequests() {
- // Set the notification state back up. If anything goes wrong,
- // disconnect the client.
- if err := c.reregisterNtfns(); err != nil {
- log.Warnf("Unable to re-establish notification state: %v", err)
- c.Disconnect()
- return
- }
-
- // Since it's possible to block on send and more requests might be
- // added by the caller while resending, make a copy of all of the
- // requests that need to be resent now and work from the copy. This
- // also allows the lock to be released quickly.
- c.requestLock.Lock()
- resendReqs := make([]*jsonRequest, 0, c.requestList.Len())
- var nextElem *list.Element
- for e := c.requestList.Front(); e != nil; e = nextElem {
- nextElem = e.Next()
-
- jReq := e.Value.(*jsonRequest)
- if _, ok := ignoreResends[jReq.method]; ok {
- // If a request is not sent on reconnect, remove it
- // from the request structures, since no reply is
- // expected.
- delete(c.requestMap, jReq.id)
- c.requestList.Remove(e)
- } else {
- resendReqs = append(resendReqs, jReq)
- }
- }
- c.requestLock.Unlock()
-
- for _, jReq := range resendReqs {
- // Stop resending commands if the client disconnected again
- // since the next reconnect will handle them.
- if c.Disconnected() {
- return
- }
-
- log.Tracef("Sending command [%s] with id %d", jReq.method,
- jReq.id)
- c.sendMessage(jReq.marshalledJSON)
- }
-}
-
-// wsReconnectHandler listens for client disconnects and automatically tries
-// to reconnect with retry interval that scales based on the number of retries.
-// It also resends any commands that had not completed when the client
-// disconnected so the disconnect/reconnect process is largely transparent to
-// the caller. This function is not run when the DisableAutoReconnect config
-// options is set.
-//
-// This function must be run as a goroutine.
-func (c *Client) wsReconnectHandler() {
-out:
- for {
- select {
- case <-c.disconnect:
- // On disconnect, fallthrough to reestablish the
- // connection.
-
- case <-c.shutdown:
- break out
- }
-
- reconnect:
- for {
- select {
- case <-c.shutdown:
- break out
- default:
- }
-
- wsConn, err := dial(c.config)
- if err != nil {
- c.retryCount++
- log.Infof("Failed to connect to %s: %v",
- c.config.Host, err)
-
- // Scale the retry interval by the number of
- // retries so there is a backoff up to a max
- // of 1 minute.
- scaledInterval := connectionRetryInterval.Nanoseconds() * c.retryCount
- scaledDuration := time.Duration(scaledInterval)
- if scaledDuration > time.Minute {
- scaledDuration = time.Minute
- }
- log.Infof("Retrying connection to %s in "+
- "%s", c.config.Host, scaledDuration)
- time.Sleep(scaledDuration)
- continue reconnect
- }
-
- log.Infof("Reestablished connection to RPC server %s",
- c.config.Host)
-
- // Reset the connection state and signal the reconnect
- // has happened.
- c.wsConn = wsConn
- c.retryCount = 0
-
- c.mtx.Lock()
- c.disconnect = make(chan struct{})
- c.disconnected = false
- c.mtx.Unlock()
-
- // Start processing input and output for the
- // new connection.
- c.start()
-
- // Reissue pending requests in another goroutine since
- // the send can block.
- go c.resendRequests()
-
- // Break out of the reconnect loop back to wait for
- // disconnect again.
- break reconnect
- }
- }
- c.wg.Done()
- log.Tracef("RPC client reconnect handler done for %s", c.config.Host)
-}
-
-// handleSendPostMessage handles performing the passed HTTP request, reading the
-// result, unmarshalling it, and delivering the unmarshalled result to the
-// provided response channel.
-func (c *Client) handleSendPostMessage(details *sendPostDetails) {
- jReq := details.jsonRequest
- log.Tracef("Sending command [%s] with id %d", jReq.method, jReq.id)
- httpResponse, err := c.httpClient.Do(details.httpRequest)
- if err != nil {
- jReq.responseChan <- &response{err: err}
- return
- }
-
- // Read the raw bytes and close the response.
- respBytes, err := ioutil.ReadAll(httpResponse.Body)
- httpResponse.Body.Close()
- if err != nil {
- err = fmt.Errorf("error reading json reply: %v", err)
- jReq.responseChan <- &response{err: err}
- return
- }
-
- // Try to unmarshal the response as a regular JSON-RPC response.
- var resp rawResponse
- err = json.Unmarshal(respBytes, &resp)
- if err != nil {
- // When the response itself isn't a valid JSON-RPC response
- // return an error which includes the HTTP status code and raw
- // response bytes.
- err = fmt.Errorf("status code: %d, response: %q",
- httpResponse.StatusCode, string(respBytes))
- jReq.responseChan <- &response{err: err}
- return
- }
-
- res, err := resp.result()
- jReq.responseChan <- &response{result: res, err: err}
-}
-
-// sendPostHandler handles all outgoing messages when the client is running
-// in HTTP POST mode. It uses a buffered channel to serialize output messages
-// while allowing the sender to continue running asynchronously. It must be run
-// as a goroutine.
-func (c *Client) sendPostHandler() {
-out:
- for {
- // Send any messages ready for send until the shutdown channel
- // is closed.
- select {
- case details := <-c.sendPostChan:
- c.handleSendPostMessage(details)
-
- case <-c.shutdown:
- break out
- }
- }
-
- // Drain any wait channels before exiting so nothing is left waiting
- // around to send.
-cleanup:
- for {
- select {
- case details := <-c.sendPostChan:
- details.jsonRequest.responseChan <- &response{
- result: nil,
- err: ErrClientShutdown,
- }
-
- default:
- break cleanup
- }
- }
- c.wg.Done()
- log.Tracef("RPC client send handler done for %s", c.config.Host)
-
-}
-
-// sendPostRequest sends the passed HTTP request to the RPC server using the
-// HTTP client associated with the client. It is backed by a buffered channel,
-// so it will not block until the send channel is full.
-func (c *Client) sendPostRequest(httpReq *http.Request, jReq *jsonRequest) {
- // Don't send the message if shutting down.
- select {
- case <-c.shutdown:
- jReq.responseChan <- &response{result: nil, err: ErrClientShutdown}
- default:
- }
-
- c.sendPostChan <- &sendPostDetails{
- jsonRequest: jReq,
- httpRequest: httpReq,
- }
-}
-
-// newFutureError returns a new future result channel that already has the
-// passed error waitin on the channel with the reply set to nil. This is useful
-// to easily return errors from the various Async functions.
-func newFutureError(err error) chan *response {
- responseChan := make(chan *response, 1)
- responseChan <- &response{err: err}
- return responseChan
-}
-
-// receiveFuture receives from the passed futureResult channel to extract a
-// reply or any errors. The examined errors include an error in the
-// futureResult and the error in the reply from the server. This will block
-// until the result is available on the passed channel.
-func receiveFuture(f chan *response) ([]byte, error) {
- // Wait for a response on the returned channel.
- r := <-f
- return r.result, r.err
-}
-
-// sendPost sends the passed request to the server by issuing an HTTP POST
-// request using the provided response channel for the reply. Typically a new
-// connection is opened and closed for each command when using this method,
-// however, the underlying HTTP client might coalesce multiple commands
-// depending on several factors including the remote server configuration.
-func (c *Client) sendPost(jReq *jsonRequest) {
- // Generate a request to the configured RPC server.
- protocol := "http"
- if !c.config.DisableTLS {
- protocol = "https"
- }
- url := protocol + "://" + c.config.Host
- bodyReader := bytes.NewReader(jReq.marshalledJSON)
- httpReq, err := http.NewRequest("POST", url, bodyReader)
- if err != nil {
- jReq.responseChan <- &response{result: nil, err: err}
- return
- }
- httpReq.Close = true
- httpReq.Header.Set("Content-Type", "application/json")
-
- // Configure basic access authorization.
- httpReq.SetBasicAuth(c.config.User, c.config.Pass)
-
- log.Tracef("Sending command [%s] with id %d", jReq.method, jReq.id)
- c.sendPostRequest(httpReq, jReq)
-}
-
-// sendRequest sends the passed json request to the associated server using the
-// provided response channel for the reply. It handles both websocket and HTTP
-// POST mode depending on the configuration of the client.
-func (c *Client) sendRequest(jReq *jsonRequest) {
- // Choose which marshal and send function to use depending on whether
- // the client running in HTTP POST mode or not. When running in HTTP
- // POST mode, the command is issued via an HTTP client. Otherwise,
- // the command is issued via the asynchronous websocket channels.
- if c.config.HTTPPostMode {
- c.sendPost(jReq)
- return
- }
-
- // Check whether the websocket connection has never been established,
- // in which case the handler goroutines are not running.
- select {
- case <-c.connEstablished:
- default:
- jReq.responseChan <- &response{err: ErrClientNotConnected}
- return
- }
-
- // Add the request to the internal tracking map so the response from the
- // remote server can be properly detected and routed to the response
- // channel. Then send the marshalled request via the websocket
- // connection.
- if err := c.addRequest(jReq); err != nil {
- jReq.responseChan <- &response{err: err}
- return
- }
- log.Tracef("Sending command [%s] with id %d", jReq.method, jReq.id)
- c.sendMessage(jReq.marshalledJSON)
-}
-
-// sendCmd sends the passed command to the associated server and returns a
-// response channel on which the reply will be delivered at some point in the
-// future. It handles both websocket and HTTP POST mode depending on the
-// configuration of the client.
-func (c *Client) sendCmd(cmd interface{}) chan *response {
- // Get the method associated with the command.
- method, err := btcjson.CmdMethod(cmd)
- if err != nil {
- return newFutureError(err)
- }
-
- // Marshal the command.
- id := c.NextID()
- marshalledJSON, err := btcjson.MarshalCmd(id, cmd)
- if err != nil {
- return newFutureError(err)
- }
-
- // Generate the request and send it along with a channel to respond on.
- responseChan := make(chan *response, 1)
- jReq := &jsonRequest{
- id: id,
- method: method,
- cmd: cmd,
- marshalledJSON: marshalledJSON,
- responseChan: responseChan,
- }
- c.sendRequest(jReq)
-
- return responseChan
-}
-
-// sendCmdAndWait sends the passed command to the associated server, waits
-// for the reply, and returns the result from it. It will return the error
-// field in the reply if there is one.
-func (c *Client) sendCmdAndWait(cmd interface{}) (interface{}, error) {
- // Marshal the command to JSON-RPC, send it to the connected server, and
- // wait for a response on the returned channel.
- return receiveFuture(c.sendCmd(cmd))
-}
-
-// Disconnected returns whether or not the server is disconnected. If a
-// websocket client was created but never connected, this also returns false.
-func (c *Client) Disconnected() bool {
- c.mtx.Lock()
- defer c.mtx.Unlock()
-
- select {
- case <-c.connEstablished:
- return c.disconnected
- default:
- return false
- }
-}
-
-// doDisconnect disconnects the websocket associated with the client if it
-// hasn't already been disconnected. It will return false if the disconnect is
-// not needed or the client is running in HTTP POST mode.
-//
-// This function is safe for concurrent access.
-func (c *Client) doDisconnect() bool {
- if c.config.HTTPPostMode {
- return false
- }
-
- c.mtx.Lock()
- defer c.mtx.Unlock()
-
- // Nothing to do if already disconnected.
- if c.disconnected {
- return false
- }
-
- log.Tracef("Disconnecting RPC client %s", c.config.Host)
- close(c.disconnect)
- if c.wsConn != nil {
- c.wsConn.Close()
- }
- c.disconnected = true
- return true
-}
-
-// doShutdown closes the shutdown channel and logs the shutdown unless shutdown
-// is already in progress. It will return false if the shutdown is not needed.
-//
-// This function is safe for concurrent access.
-func (c *Client) doShutdown() bool {
- // Ignore the shutdown request if the client is already in the process
- // of shutting down or already shutdown.
- select {
- case <-c.shutdown:
- return false
- default:
- }
-
- log.Tracef("Shutting down RPC client %s", c.config.Host)
- close(c.shutdown)
- return true
-}
-
-// Disconnect disconnects the current websocket associated with the client. The
-// connection will automatically be re-established unless the client was
-// created with the DisableAutoReconnect flag.
-//
-// This function has no effect when the client is running in HTTP POST mode.
-func (c *Client) Disconnect() {
- // Nothing to do if already disconnected or running in HTTP POST mode.
- if !c.doDisconnect() {
- return
- }
-
- c.requestLock.Lock()
- defer c.requestLock.Unlock()
-
- // When operating without auto reconnect, send errors to any pending
- // requests and shutdown the client.
- if c.config.DisableAutoReconnect {
- for e := c.requestList.Front(); e != nil; e = e.Next() {
- req := e.Value.(*jsonRequest)
- req.responseChan <- &response{
- result: nil,
- err: ErrClientDisconnect,
- }
- }
- c.removeAllRequests()
- c.doShutdown()
- }
-}
-
-// Shutdown shuts down the client by disconnecting any connections associated
-// with the client and, when automatic reconnect is enabled, preventing future
-// attempts to reconnect. It also stops all goroutines.
-func (c *Client) Shutdown() {
- // Do the shutdown under the request lock to prevent clients from
- // adding new requests while the client shutdown process is initiated.
- c.requestLock.Lock()
- defer c.requestLock.Unlock()
-
- // Ignore the shutdown request if the client is already in the process
- // of shutting down or already shutdown.
- if !c.doShutdown() {
- return
- }
-
- // Send the ErrClientShutdown error to any pending requests.
- for e := c.requestList.Front(); e != nil; e = e.Next() {
- req := e.Value.(*jsonRequest)
- req.responseChan <- &response{
- result: nil,
- err: ErrClientShutdown,
- }
- }
- c.removeAllRequests()
-
- // Disconnect the client if needed.
- c.doDisconnect()
-}
-
-// start begins processing input and output messages.
-func (c *Client) start() {
- log.Tracef("Starting RPC client %s", c.config.Host)
-
- // Start the I/O processing handlers depending on whether the client is
- // in HTTP POST mode or the default websocket mode.
- if c.config.HTTPPostMode {
- c.wg.Add(1)
- go c.sendPostHandler()
- } else {
- c.wg.Add(3)
- go func() {
- if c.ntfnHandlers != nil {
- if c.ntfnHandlers.OnClientConnected != nil {
- c.ntfnHandlers.OnClientConnected()
- }
- }
- c.wg.Done()
- }()
- go c.wsInHandler()
- go c.wsOutHandler()
- }
-}
-
-// WaitForShutdown blocks until the client goroutines are stopped and the
-// connection is closed.
-func (c *Client) WaitForShutdown() {
- c.wg.Wait()
-}
-
-// ConnConfig describes the connection configuration parameters for the client.
-// This
-type ConnConfig struct {
- // Host is the IP address and port of the RPC server you want to connect
- // to.
- Host string
-
- // Endpoint is the websocket endpoint on the RPC server. This is
- // typically "ws".
- Endpoint string
-
- // User is the username to use to authenticate to the RPC server.
- User string
-
- // Pass is the passphrase to use to authenticate to the RPC server.
- Pass string
-
- // DisableTLS specifies whether transport layer security should be
- // disabled. It is recommended to always use TLS if the RPC server
- // supports it as otherwise your username and password is sent across
- // the wire in cleartext.
- DisableTLS bool
-
- // Certificates are the bytes for a PEM-encoded certificate chain used
- // for the TLS connection. It has no effect if the DisableTLS parameter
- // is true.
- Certificates []byte
-
- // Proxy specifies to connect through a SOCKS 5 proxy server. It may
- // be an empty string if a proxy is not required.
- Proxy string
-
- // ProxyUser is an optional username to use for the proxy server if it
- // requires authentication. It has no effect if the Proxy parameter
- // is not set.
- ProxyUser string
-
- // ProxyPass is an optional password to use for the proxy server if it
- // requires authentication. It has no effect if the Proxy parameter
- // is not set.
- ProxyPass string
-
- // DisableAutoReconnect specifies the client should not automatically
- // try to reconnect to the server when it has been disconnected.
- DisableAutoReconnect bool
-
- // DisableConnectOnNew specifies that a websocket client connection
- // should not be tried when creating the client with New. Instead, the
- // client is created and returned unconnected, and Connect must be
- // called manually.
- DisableConnectOnNew bool
-
- // HTTPPostMode instructs the client to run using multiple independent
- // connections issuing HTTP POST requests instead of using the default
- // of websockets. Websockets are generally preferred as some of the
- // features of the client such notifications only work with websockets,
- // however, not all servers support the websocket extensions, so this
- // flag can be set to true to use basic HTTP POST requests instead.
- HTTPPostMode bool
-
- // EnableBCInfoHacks is an option provided to enable compatiblity hacks
- // when connecting to blockchain.info RPC server
- EnableBCInfoHacks bool
-}
-
-// newHTTPClient returns a new http client that is configured according to the
-// proxy and TLS settings in the associated connection configuration.
-func newHTTPClient(config *ConnConfig) (*http.Client, error) {
- // Set proxy function if there is a proxy configured.
- var proxyFunc func(*http.Request) (*url.URL, error)
- if config.Proxy != "" {
- proxyURL, err := url.Parse(config.Proxy)
- if err != nil {
- return nil, err
- }
- proxyFunc = http.ProxyURL(proxyURL)
- }
-
- // Configure TLS if needed.
- var tlsConfig *tls.Config
- if !config.DisableTLS {
- if len(config.Certificates) > 0 {
- pool := x509.NewCertPool()
- pool.AppendCertsFromPEM(config.Certificates)
- tlsConfig = &tls.Config{
- RootCAs: pool,
- }
- }
- }
-
- client := http.Client{
- Transport: &http.Transport{
- Proxy: proxyFunc,
- TLSClientConfig: tlsConfig,
- },
- }
-
- return &client, nil
-}
-
-// dial opens a websocket connection using the passed connection configuration
-// details.
-func dial(config *ConnConfig) (*websocket.Conn, error) {
- // Setup TLS if not disabled.
- var tlsConfig *tls.Config
- var scheme = "ws"
- if !config.DisableTLS {
- tlsConfig = &tls.Config{
- MinVersion: tls.VersionTLS12,
- }
- if len(config.Certificates) > 0 {
- pool := x509.NewCertPool()
- pool.AppendCertsFromPEM(config.Certificates)
- tlsConfig.RootCAs = pool
- }
- scheme = "wss"
- }
-
- // Create a websocket dialer that will be used to make the connection.
- // It is modified by the proxy setting below as needed.
- dialer := websocket.Dialer{TLSClientConfig: tlsConfig}
-
- // Setup the proxy if one is configured.
- if config.Proxy != "" {
- proxy := &socks.Proxy{
- Addr: config.Proxy,
- Username: config.ProxyUser,
- Password: config.ProxyPass,
- }
- dialer.NetDial = proxy.Dial
- }
-
- // The RPC server requires basic authorization, so create a custom
- // request header with the Authorization header set.
- login := config.User + ":" + config.Pass
- auth := "Basic " + base64.StdEncoding.EncodeToString([]byte(login))
- requestHeader := make(http.Header)
- requestHeader.Add("Authorization", auth)
-
- // Dial the connection.
- url := fmt.Sprintf("%s://%s/%s", scheme, config.Host, config.Endpoint)
- wsConn, resp, err := dialer.Dial(url, requestHeader)
- if err != nil {
- if err != websocket.ErrBadHandshake || resp == nil {
- return nil, err
- }
-
- // Detect HTTP authentication error status codes.
- if resp.StatusCode == http.StatusUnauthorized ||
- resp.StatusCode == http.StatusForbidden {
- return nil, ErrInvalidAuth
- }
-
- // The connection was authenticated and the status response was
- // ok, but the websocket handshake still failed, so the endpoint
- // is invalid in some way.
- if resp.StatusCode == http.StatusOK {
- return nil, ErrInvalidEndpoint
- }
-
- // Return the status text from the server if none of the special
- // cases above apply.
- return nil, errors.New(resp.Status)
- }
- return wsConn, nil
-}
-
-// New creates a new RPC client based on the provided connection configuration
-// details. The notification handlers parameter may be nil if you are not
-// interested in receiving notifications and will be ignored if the
-// configuration is set to run in HTTP POST mode.
-func New(config *ConnConfig, ntfnHandlers *NotificationHandlers) (*Client, error) {
- // Either open a websocket connection or create an HTTP client depending
- // on the HTTP POST mode. Also, set the notification handlers to nil
- // when running in HTTP POST mode.
- var wsConn *websocket.Conn
- var httpClient *http.Client
- connEstablished := make(chan struct{})
- var start bool
- if config.HTTPPostMode {
- ntfnHandlers = nil
- start = true
-
- var err error
- httpClient, err = newHTTPClient(config)
- if err != nil {
- return nil, err
- }
- } else {
- if !config.DisableConnectOnNew {
- var err error
- wsConn, err = dial(config)
- if err != nil {
- return nil, err
- }
- start = true
- }
- }
-
- client := &Client{
- config: config,
- wsConn: wsConn,
- httpClient: httpClient,
- requestMap: make(map[uint64]*list.Element),
- requestList: list.New(),
- ntfnHandlers: ntfnHandlers,
- ntfnState: newNotificationState(),
- sendChan: make(chan []byte, sendBufferSize),
- sendPostChan: make(chan *sendPostDetails, sendPostBufferSize),
- connEstablished: connEstablished,
- disconnect: make(chan struct{}),
- shutdown: make(chan struct{}),
- }
-
- if start {
- log.Infof("Established connection to RPC server %s",
- config.Host)
- close(connEstablished)
- client.start()
- if !client.config.HTTPPostMode && !client.config.DisableAutoReconnect {
- client.wg.Add(1)
- go client.wsReconnectHandler()
- }
- }
-
- return client, nil
-}
-
-// Connect establishes the initial websocket connection. This is necessary when
-// a client was created after setting the DisableConnectOnNew field of the
-// Config struct.
-//
-// Up to tries number of connections (each after an increasing backoff) will
-// be tried if the connection can not be established. The special value of 0
-// indicates an unlimited number of connection attempts.
-//
-// This method will error if the client is not configured for websockets, if the
-// connection has already been established, or if none of the connection
-// attempts were successful.
-func (c *Client) Connect(tries int) error {
- c.mtx.Lock()
- defer c.mtx.Unlock()
-
- if c.config.HTTPPostMode {
- return ErrNotWebsocketClient
- }
- if c.wsConn != nil {
- return ErrClientAlreadyConnected
- }
-
- // Begin connection attempts. Increase the backoff after each failed
- // attempt, up to a maximum of one minute.
- var err error
- var backoff time.Duration
- for i := 0; tries == 0 || i < tries; i++ {
- var wsConn *websocket.Conn
- wsConn, err = dial(c.config)
- if err != nil {
- backoff = connectionRetryInterval * time.Duration(i+1)
- if backoff > time.Minute {
- backoff = time.Minute
- }
- time.Sleep(backoff)
- continue
- }
-
- // Connection was established. Set the websocket connection
- // member of the client and start the goroutines necessary
- // to run the client.
- log.Infof("Established connection to RPC server %s",
- c.config.Host)
- c.wsConn = wsConn
- close(c.connEstablished)
- c.start()
- if !c.config.DisableAutoReconnect {
- c.wg.Add(1)
- go c.wsReconnectHandler()
- }
- return nil
- }
-
- // All connection attempts failed, so return the last error.
- return err
-}
OSDN Git Service
