// Copyright 2015, 2016 Eris Industries (UK) Ltd. // This file is part of Eris-RT // Eris-RT is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // Eris-RT is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU General Public License for more details. // You should have received a copy of the GNU General Public License // along with Eris-RT. If not, see <http://www.gnu.org/licenses/>. package client import ( "bytes" "fmt" "time" "github.com/tendermint/go-rpc/client" "github.com/tendermint/go-wire" "github.com/eris-ltd/eris-db/logging" "github.com/eris-ltd/eris-db/logging/loggers" ctypes "github.com/eris-ltd/eris-db/rpc/tendermint/core/types" "github.com/eris-ltd/eris-db/txs" ) const ( MaxCommitWaitTimeSeconds = 10 ) type Confirmation struct { BlockHash []byte Event txs.EventData Exception error Error error } // NOTE [ben] Compiler check to ensure erisNodeClient successfully implements // eris-db/client.NodeClient var _ NodeWebsocketClient = (*erisNodeWebsocketClient)(nil) type erisNodeWebsocketClient struct { // TODO: assert no memory leak on closing with open websocket tendermintWebsocket *rpcclient.WSClient logger loggers.InfoTraceLogger } // Subscribe to an eventid func (erisNodeWebsocketClient *erisNodeWebsocketClient) Subscribe(eventid string) error { // TODO we can in the background listen to the subscription id and remember it to ease unsubscribing later. return erisNodeWebsocketClient.tendermintWebsocket.Subscribe(eventid) } // Unsubscribe from an eventid func (erisNodeWebsocketClient *erisNodeWebsocketClient) Unsubscribe(subscriptionId string) error { return erisNodeWebsocketClient.tendermintWebsocket.Unsubscribe(subscriptionId) } // Returns a channel that will receive a confirmation with a result or the exception that // has been confirmed; or an error is returned and the confirmation channel is nil. func (erisNodeWebsocketClient *erisNodeWebsocketClient) WaitForConfirmation(tx txs.Tx, chainId string, inputAddr []byte) (chan Confirmation, error) { // check no errors are reported on the websocket if err := erisNodeWebsocketClient.assertNoErrors(); err != nil { return nil, err } // Setup the confirmation channel to be returned confirmationChannel := make(chan Confirmation, 1) var latestBlockHash []byte eid := txs.EventStringAccInput(inputAddr) if err := erisNodeWebsocketClient.tendermintWebsocket.Subscribe(eid); err != nil { return nil, fmt.Errorf("Error subscribing to AccInput event (%s): %v", eid, err) } if err := erisNodeWebsocketClient.tendermintWebsocket.Subscribe(txs.EventStringNewBlock()); err != nil { return nil, fmt.Errorf("Error subscribing to NewBlock event: %v", err) } // Read the incoming events go func() { var err error for { resultBytes := <-erisNodeWebsocketClient.tendermintWebsocket.ResultsCh result := new(ctypes.ErisDBResult) if wire.ReadJSONPtr(result, resultBytes, &err); err != nil { // keep calm and carry on logging.InfoMsg(erisNodeWebsocketClient.logger, "Failed to unmarshal json bytes for websocket event", "error", err) continue } subscription, ok := (*result).(*ctypes.ResultSubscribe) if ok { // Received confirmation of subscription to event streams // TODO: collect subscription IDs, push into channel and on completion // unsubscribe logging.InfoMsg(erisNodeWebsocketClient.logger, "Received confirmation for event", "event", subscription.Event, "subscription_id", subscription.SubscriptionId) continue } event, ok := (*result).(*ctypes.ResultEvent) if !ok { // keep calm and carry on logging.InfoMsg(erisNodeWebsocketClient.logger, "Failed to cast to ResultEvent for websocket event", "event", event.Event) continue } blockData, ok := event.Data.(txs.EventDataNewBlock) if ok { latestBlockHash = blockData.Block.Hash() logging.TraceMsg(erisNodeWebsocketClient.logger, "Registered new block", "block", blockData.Block, "latest_block_hash", latestBlockHash, ) continue } // we don't accept events unless they came after a new block (ie. in) if latestBlockHash == nil { logging.InfoMsg(erisNodeWebsocketClient.logger, "First block has not been registered so ignoring event", "event", event.Event) continue } if event.Event != eid { logging.InfoMsg(erisNodeWebsocketClient.logger, "Received unsolicited event", "event_received", event.Event, "event_expected", eid) continue } data, ok := event.Data.(txs.EventDataTx) if !ok { // We are on the lookout for EventDataTx confirmationChannel <- Confirmation{ BlockHash: latestBlockHash, Event: nil, Exception: fmt.Errorf("response error: expected result.Data to be *types.EventDataTx"), Error: nil, } return } if !bytes.Equal(txs.TxHash(chainId, data.Tx), txs.TxHash(chainId, tx)) { logging.TraceMsg(erisNodeWebsocketClient.logger, "Received different event", // TODO: consider re-implementing TxID again, or other more clear debug "received transaction event", txs.TxHash(chainId, data.Tx)) continue } if data.Exception != "" { confirmationChannel <- Confirmation{ BlockHash: latestBlockHash, Event: &data, Exception: fmt.Errorf("Transaction confirmed with exception:", data.Exception), Error: nil, } return } // success, return the full event and blockhash and exit go-routine confirmationChannel <- Confirmation{ BlockHash: latestBlockHash, Event: &data, Exception: nil, Error: nil, } return } }() // TODO: [ben] this is a draft implementation as resources on time.After can not be // recovered before the timeout. Close-down timeout at success properly. timeout := time.After(time.Duration(MaxCommitWaitTimeSeconds) * time.Second) go func() { <-timeout confirmationChannel <- Confirmation{ BlockHash: nil, Event: nil, Exception: nil, Error: fmt.Errorf("timed out waiting for event"), } return }() return confirmationChannel, nil } func (erisNodeWebsocketClient *erisNodeWebsocketClient) Close() { if erisNodeWebsocketClient.tendermintWebsocket != nil { erisNodeWebsocketClient.tendermintWebsocket.Stop() } } func (erisNodeWebsocketClient *erisNodeWebsocketClient) assertNoErrors() error { if erisNodeWebsocketClient.tendermintWebsocket != nil { select { case err := <-erisNodeWebsocketClient.tendermintWebsocket.ErrorsCh: return err default: return nil } } else { return fmt.Errorf("Eris-client has no websocket initialised.") } }