diff --git a/ethcore/light/src/net/context.rs b/ethcore/light/src/net/context.rs index e58bca7f3..d9910f958 100644 --- a/ethcore/light/src/net/context.rs +++ b/ethcore/light/src/net/context.rs @@ -16,13 +16,13 @@ //! I/O and event context generalizations. -use network::{NetworkContext, PeerId}; +use network::{NetworkContext, PeerId, NodeId}; use super::{Announcement, LightProtocol, ReqId}; use super::error::Error; use request::Request; -/// An I/O context which allows sending and receiving packets as well as +/// An I/O context which allows sending and receiving packets as well as /// disconnecting peers. This is used as a generalization of the portions /// of a p2p network which the light protocol structure makes use of. pub trait IoContext { @@ -41,8 +41,12 @@ pub trait IoContext { /// Get a peer's protocol version. fn protocol_version(&self, peer: PeerId) -> Option; + + /// Persistent peer id + fn persistent_peer_id(&self, peer: PeerId) -> Option; } + impl<'a> IoContext for NetworkContext<'a> { fn send(&self, peer: PeerId, packet_id: u8, packet_body: Vec) { if let Err(e) = self.send(peer, packet_id, packet_body) { @@ -67,6 +71,10 @@ impl<'a> IoContext for NetworkContext<'a> { fn protocol_version(&self, peer: PeerId) -> Option { self.protocol_version(self.subprotocol_name(), peer) } + + fn persistent_peer_id(&self, peer: PeerId) -> Option { + self.session_info(peer).and_then(|info| info.id) + } } /// Context for a protocol event. @@ -75,6 +83,9 @@ pub trait EventContext { /// disconnected/connected peer. fn peer(&self) -> PeerId; + /// Returns the relevant's peer persistent Id (aka NodeId). + fn persistent_peer_id(&self, peer: PeerId) -> Option; + /// Make a request from a peer. fn request_from(&self, peer: PeerId, request: Request) -> Result; @@ -89,7 +100,7 @@ pub trait EventContext { fn disable_peer(&self, peer: PeerId); } -/// Concrete implementation of `EventContext` over the light protocol struct and +/// Concrete implementation of `EventContext` over the light protocol struct and /// an io context. pub struct Ctx<'a> { /// Io context to enable immediate response to events. @@ -97,11 +108,18 @@ pub struct Ctx<'a> { /// Protocol implementation. pub proto: &'a LightProtocol, /// Relevant peer for event. - pub peer: PeerId, + pub peer: PeerId, } impl<'a> EventContext for Ctx<'a> { - fn peer(&self) -> PeerId { self.peer } + fn peer(&self) -> PeerId { + self.peer + } + + fn persistent_peer_id(&self, id: PeerId) -> Option { + self.io.persistent_peer_id(id) + } + fn request_from(&self, peer: PeerId, request: Request) -> Result { self.proto.request_from(self.io, &peer, request) } @@ -117,4 +135,4 @@ impl<'a> EventContext for Ctx<'a> { fn disable_peer(&self, peer: PeerId) { self.io.disable_peer(peer); } -} \ No newline at end of file +} diff --git a/ethcore/light/src/net/tests/mod.rs b/ethcore/light/src/net/tests/mod.rs index 24ce4250a..99f695b81 100644 --- a/ethcore/light/src/net/tests/mod.rs +++ b/ethcore/light/src/net/tests/mod.rs @@ -15,13 +15,13 @@ // along with Parity. If not, see . //! Tests for the `LightProtocol` implementation. -//! These don't test of the higher level logic on top of +//! These don't test of the higher level logic on top of use ethcore::blockchain_info::BlockChainInfo; use ethcore::client::{BlockChainClient, EachBlockWith, TestBlockChainClient}; use ethcore::ids::BlockId; use ethcore::transaction::SignedTransaction; -use network::PeerId; +use network::{PeerId, NodeId}; use net::buffer_flow::FlowParams; use net::context::IoContext; @@ -68,6 +68,10 @@ impl IoContext for Expect { fn protocol_version(&self, _peer: PeerId) -> Option { Some(super::MAX_PROTOCOL_VERSION) } + + fn persistent_peer_id(&self, _peer: PeerId) -> Option { + None + } } // can't implement directly for Arc due to cross-crate orphan rules. @@ -117,7 +121,7 @@ impl Provider for TestProvider { .map(|x: u64| x.saturating_mul(req.skip + 1)) .take_while(|x| if req.reverse { x < &start_num } else { best_num - start_num >= *x }) .map(|x| if req.reverse { start_num - x } else { start_num + x }) - .map(|x| self.0.client.block_header(BlockId::Number(x))) + .map(|x| self.0.client.block_header(BlockId::Number(x))) .take_while(|x| x.is_some()) .flat_map(|x| x) .collect() @@ -150,12 +154,12 @@ impl Provider for TestProvider { } } }) - .collect() + .collect() } fn contract_code(&self, req: request::ContractCodes) -> Vec { req.code_requests.into_iter() - .map(|req| { + .map(|req| { req.account_key.iter().chain(req.account_key.iter()).cloned().collect() }) .collect() @@ -213,9 +217,9 @@ fn status(chain_info: BlockChainInfo) -> Status { #[test] fn handshake_expected() { let flow_params = make_flow_params(); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let status = status(provider.client.chain_info()); @@ -228,9 +232,9 @@ fn handshake_expected() { #[should_panic] fn genesis_mismatch() { let flow_params = make_flow_params(); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let mut status = status(provider.client.chain_info()); status.genesis_hash = H256::default(); @@ -243,15 +247,15 @@ fn genesis_mismatch() { #[test] fn buffer_overflow() { let flow_params = make_flow_params(); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let status = status(provider.client.chain_info()); { - let packet_body = write_handshake(&status, &capabilities, Some(&flow_params)); - proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); + let packet_body = write_handshake(&status, &capabilities, Some(&flow_params)); + proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); } { @@ -276,9 +280,9 @@ fn buffer_overflow() { #[test] fn get_block_headers() { let flow_params = FlowParams::new(5_000_000.into(), Default::default(), 0.into()); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let cur_status = status(provider.client.chain_info()); let my_status = write_handshake(&cur_status, &capabilities, Some(&flow_params)); @@ -288,8 +292,8 @@ fn get_block_headers() { let cur_status = status(provider.client.chain_info()); { - let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); - proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); + let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); + proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); proto.handle_packet(&Expect::Nothing, &1, packet::STATUS, &my_status); } @@ -309,7 +313,7 @@ fn get_block_headers() { let new_buf = *flow_params.limit() - flow_params.compute_cost(request::Kind::Headers, 10); let mut response_stream = RlpStream::new_list(3); - + response_stream.append(&req_id).append(&new_buf).begin_list(10); for header in headers { response_stream.append_raw(&header, 1); @@ -325,9 +329,9 @@ fn get_block_headers() { #[test] fn get_block_bodies() { let flow_params = FlowParams::new(5_000_000.into(), Default::default(), 0.into()); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let cur_status = status(provider.client.chain_info()); let my_status = write_handshake(&cur_status, &capabilities, Some(&flow_params)); @@ -337,8 +341,8 @@ fn get_block_bodies() { let cur_status = status(provider.client.chain_info()); { - let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); - proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); + let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); + proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); proto.handle_packet(&Expect::Nothing, &1, packet::STATUS, &my_status); } @@ -356,7 +360,7 @@ fn get_block_bodies() { let new_buf = *flow_params.limit() - flow_params.compute_cost(request::Kind::Bodies, 10); let mut response_stream = RlpStream::new_list(3); - + response_stream.append(&req_id).append(&new_buf).begin_list(10); for body in bodies { response_stream.append_raw(&body, 1); @@ -372,9 +376,9 @@ fn get_block_bodies() { #[test] fn get_block_receipts() { let flow_params = FlowParams::new(5_000_000.into(), Default::default(), 0.into()); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let cur_status = status(provider.client.chain_info()); let my_status = write_handshake(&cur_status, &capabilities, Some(&flow_params)); @@ -384,8 +388,8 @@ fn get_block_receipts() { let cur_status = status(provider.client.chain_info()); { - let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); - proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); + let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); + proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body)); proto.handle_packet(&Expect::Nothing, &1, packet::STATUS, &my_status); } @@ -409,7 +413,7 @@ fn get_block_receipts() { let new_buf = *flow_params.limit() - flow_params.compute_cost(request::Kind::Receipts, receipts.len()); let mut response_stream = RlpStream::new_list(3); - + response_stream.append(&req_id).append(&new_buf).begin_list(receipts.len()); for block_receipts in receipts { response_stream.append_raw(&block_receipts, 1); @@ -425,15 +429,15 @@ fn get_block_receipts() { #[test] fn get_state_proofs() { let flow_params = FlowParams::new(5_000_000.into(), Default::default(), 0.into()); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let cur_status = status(provider.client.chain_info()); { - let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); - proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body.clone())); + let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); + proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body.clone())); proto.handle_packet(&Expect::Nothing, &1, packet::STATUS, &packet_body); } @@ -441,7 +445,7 @@ fn get_state_proofs() { let key1 = U256::from(11223344).into(); let key2 = U256::from(99988887).into(); - let request = Request::StateProofs (request::StateProofs { + let request = Request::StateProofs (request::StateProofs { requests: vec![ request::StateProof { block: H256::default(), key1: key1, key2: None, from_level: 0 }, request::StateProof { block: H256::default(), key1: key1, key2: Some(key2), from_level: 0}, @@ -458,7 +462,7 @@ fn get_state_proofs() { let new_buf = *flow_params.limit() - flow_params.compute_cost(request::Kind::StateProofs, 2); let mut response_stream = RlpStream::new_list(3); - + response_stream.append(&req_id).append(&new_buf).begin_list(2); for proof in proofs { response_stream.append_raw(&proof, 1); @@ -474,15 +478,15 @@ fn get_state_proofs() { #[test] fn get_contract_code() { let flow_params = FlowParams::new(5_000_000.into(), Default::default(), 0.into()); - let capabilities = capabilities(); + let capabilities = capabilities(); - let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); + let (provider, proto) = setup(flow_params.clone(), capabilities.clone()); let cur_status = status(provider.client.chain_info()); { - let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); - proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body.clone())); + let packet_body = write_handshake(&cur_status, &capabilities, Some(&flow_params)); + proto.on_connect(&1, &Expect::Send(1, packet::STATUS, packet_body.clone())); proto.handle_packet(&Expect::Nothing, &1, packet::STATUS, &packet_body); } @@ -490,7 +494,7 @@ fn get_contract_code() { let key1 = U256::from(11223344).into(); let key2 = U256::from(99988887).into(); - let request = Request::Codes (request::ContractCodes { + let request = Request::Codes (request::ContractCodes { code_requests: vec![ request::ContractCode { block_hash: H256::default(), account_key: key1 }, request::ContractCode { block_hash: H256::default(), account_key: key2 }, @@ -507,7 +511,7 @@ fn get_contract_code() { let new_buf = *flow_params.limit() - flow_params.compute_cost(request::Kind::Codes, 2); let mut response_stream = RlpStream::new_list(3); - + response_stream.append(&req_id).append(&new_buf).begin_list(2); for code in codes { response_stream.append(&code); @@ -518,4 +522,4 @@ fn get_contract_code() { let expected = Expect::Respond(packet::CONTRACT_CODES, response); proto.handle_packet(&expected, &1, packet::GET_CONTRACT_CODES, &request_body); -} \ No newline at end of file +} diff --git a/ethcore/src/client/chain_notify.rs b/ethcore/src/client/chain_notify.rs index d9ebac299..a9f8d0204 100644 --- a/ethcore/src/client/chain_notify.rs +++ b/ethcore/src/client/chain_notify.rs @@ -40,6 +40,14 @@ pub trait ChainNotify : Send + Sync { fn stop(&self) { // does nothing by default } + + /// fires when new transactions are received from a peer + fn transactions_received(&self, + _hashes: Vec, + _peer_id: usize, + ) { + // does nothing by default + } } impl IpcConfig for ChainNotify { } diff --git a/ethcore/src/client/client.rs b/ethcore/src/client/client.rs index fb8b5a9f2..dce594617 100644 --- a/ethcore/src/client/client.rs +++ b/ethcore/src/client/client.rs @@ -559,11 +559,15 @@ impl Client { } /// Import transactions from the IO queue - pub fn import_queued_transactions(&self, transactions: &[Bytes]) -> usize { + pub fn import_queued_transactions(&self, transactions: &[Bytes], peer_id: usize) -> usize { trace!(target: "external_tx", "Importing queued"); let _timer = PerfTimer::new("import_queued_transactions"); self.queue_transactions.fetch_sub(transactions.len(), AtomicOrdering::SeqCst); - let txs = transactions.iter().filter_map(|bytes| UntrustedRlp::new(bytes).as_val().ok()).collect(); + let txs: Vec = transactions.iter().filter_map(|bytes| UntrustedRlp::new(bytes).as_val().ok()).collect(); + let hashes: Vec<_> = txs.iter().map(|tx| tx.hash()).collect(); + self.notify(|notify| { + notify.transactions_received(hashes.clone(), peer_id); + }); let results = self.miner.import_external_transactions(self, txs); results.len() } @@ -1264,14 +1268,14 @@ impl BlockChainClient for Client { (*self.build_last_hashes(self.chain.read().best_block_hash())).clone() } - fn queue_transactions(&self, transactions: Vec) { + fn queue_transactions(&self, transactions: Vec, peer_id: usize) { let queue_size = self.queue_transactions.load(AtomicOrdering::Relaxed); trace!(target: "external_tx", "Queue size: {}", queue_size); if queue_size > MAX_TX_QUEUE_SIZE { debug!("Ignoring {} transactions: queue is full", transactions.len()); } else { let len = transactions.len(); - match self.io_channel.lock().send(ClientIoMessage::NewTransactions(transactions)) { + match self.io_channel.lock().send(ClientIoMessage::NewTransactions(transactions, peer_id)) { Ok(_) => { self.queue_transactions.fetch_add(len, AtomicOrdering::SeqCst); } diff --git a/ethcore/src/client/test_client.rs b/ethcore/src/client/test_client.rs index 8442d723b..44954f99d 100644 --- a/ethcore/src/client/test_client.rs +++ b/ethcore/src/client/test_client.rs @@ -657,7 +657,7 @@ impl BlockChainClient for TestBlockChainClient { unimplemented!(); } - fn queue_transactions(&self, transactions: Vec) { + fn queue_transactions(&self, transactions: Vec, _peer_id: usize) { // import right here let txs = transactions.into_iter().filter_map(|bytes| UntrustedRlp::new(&bytes).as_val().ok()).collect(); self.miner.import_external_transactions(self, txs); diff --git a/ethcore/src/client/traits.rs b/ethcore/src/client/traits.rs index 4b6609cf6..f44a4496f 100644 --- a/ethcore/src/client/traits.rs +++ b/ethcore/src/client/traits.rs @@ -200,7 +200,7 @@ pub trait BlockChainClient : Sync + Send { fn last_hashes(&self) -> LastHashes; /// Queue transactions for importing. - fn queue_transactions(&self, transactions: Vec); + fn queue_transactions(&self, transactions: Vec, peer_id: usize); /// list all transactions fn pending_transactions(&self) -> Vec; @@ -294,9 +294,9 @@ pub trait ProvingBlockChainClient: BlockChainClient { /// The key is the keccak hash of the account's address. /// Returns a vector of raw trie nodes (in order from the root) proving the query. /// Nodes after `from_level` may be omitted. - /// An empty vector indicates unservable query. + /// An empty vector indicates unservable query. fn prove_account(&self, key1: H256, from_level: u32, id: BlockId) -> Vec; /// Get code by address hash. fn code_by_hash(&self, account_key: H256, id: BlockId) -> Bytes; -} \ No newline at end of file +} diff --git a/ethcore/src/service.rs b/ethcore/src/service.rs index bf346e707..d809de51a 100644 --- a/ethcore/src/service.rs +++ b/ethcore/src/service.rs @@ -39,7 +39,7 @@ pub enum ClientIoMessage { /// A block is ready BlockVerified, /// New transaction RLPs are ready to be imported - NewTransactions(Vec), + NewTransactions(Vec, usize), /// Begin snapshot restoration BeginRestoration(ManifestData), /// Feed a state chunk to the snapshot service @@ -196,7 +196,9 @@ impl IoHandler for ClientIoHandler { match *net_message { ClientIoMessage::BlockVerified => { self.client.import_verified_blocks(); } - ClientIoMessage::NewTransactions(ref transactions) => { self.client.import_queued_transactions(transactions); } + ClientIoMessage::NewTransactions(ref transactions, peer_id) => { + self.client.import_queued_transactions(transactions, peer_id); + } ClientIoMessage::BeginRestoration(ref manifest) => { if let Err(e) = self.snapshot.init_restore(manifest.clone(), true) { warn!("Failed to initialize snapshot restoration: {}", e); diff --git a/js/src/dapps/localtx/Transaction/transaction.js b/js/src/dapps/localtx/Transaction/transaction.js index fab5bf152..1756597ab 100644 --- a/js/src/dapps/localtx/Transaction/transaction.js +++ b/js/src/dapps/localtx/Transaction/transaction.js @@ -48,7 +48,6 @@ class BaseTransaction extends Component { - 0x{ transaction.nonce.toString(16) } ); } diff --git a/js/src/dapps/localtx/Transaction/transaction.spec.js b/js/src/dapps/localtx/Transaction/transaction.spec.js index 70fb4a08b..1f176a3c3 100644 --- a/js/src/dapps/localtx/Transaction/transaction.spec.js +++ b/js/src/dapps/localtx/Transaction/transaction.spec.js @@ -34,7 +34,7 @@ describe('dapps/localtx/Transaction', () => { it('renders without crashing', () => { const transaction = { hash: '0x1234567890', - nonce: 15, + nonce: new BigNumber(15), gasPrice: new BigNumber(10), gas: new BigNumber(10) }; diff --git a/rpc/src/v1/tests/helpers/sync_provider.rs b/rpc/src/v1/tests/helpers/sync_provider.rs index 039ca4c3d..af87cd98c 100644 --- a/rpc/src/v1/tests/helpers/sync_provider.rs +++ b/rpc/src/v1/tests/helpers/sync_provider.rs @@ -105,13 +105,13 @@ impl SyncProvider for TestSyncProvider { first_seen: 10, propagated_to: map![ 128.into() => 16 - ] + ], }, 5.into() => TransactionStats { first_seen: 16, propagated_to: map![ 16.into() => 1 - ] + ], } ] } diff --git a/rpc/src/v1/types/sync.rs b/rpc/src/v1/types/sync.rs index 3e40ecb85..837b425d7 100644 --- a/rpc/src/v1/types/sync.rs +++ b/rpc/src/v1/types/sync.rs @@ -157,7 +157,7 @@ impl From for TransactionStats { propagated_to: s.propagated_to .into_iter() .map(|(id, count)| (id.into(), count)) - .collect() + .collect(), } } } @@ -208,7 +208,7 @@ mod tests { first_seen: 100, propagated_to: map![ 10.into() => 50 - ] + ], }; let serialized = serde_json::to_string(&stats).unwrap(); diff --git a/sync/src/api.rs b/sync/src/api.rs index b4a45c74d..8f03fbac8 100644 --- a/sync/src/api.rs +++ b/sync/src/api.rs @@ -144,7 +144,7 @@ pub struct EthSync { network: NetworkService, /// Main (eth/par) protocol handler sync_handler: Arc, - /// Light (les) protocol handler + /// Light (les) protocol handler light_proto: Option>, /// The main subprotocol name subprotocol_name: [u8; 3], @@ -155,7 +155,7 @@ pub struct EthSync { impl EthSync { /// Creates and register protocol with the network service pub fn new(params: Params) -> Result, NetworkError> { - let pruning_info = params.chain.pruning_info(); + let pruning_info = params.chain.pruning_info(); let light_proto = match params.config.serve_light { false => None, true => Some({ @@ -297,7 +297,7 @@ impl ChainNotify for EthSync { Some(lp) => lp, None => return, }; - + let chain_info = self.sync_handler.chain.chain_info(); light_proto.make_announcement(context, Announcement { head_hash: chain_info.best_block_hash, @@ -323,7 +323,7 @@ impl ChainNotify for EthSync { // register the warp sync subprotocol self.network.register_protocol(self.sync_handler.clone(), WARP_SYNC_PROTOCOL_ID, SNAPSHOT_SYNC_PACKET_COUNT, &[1u8]) .unwrap_or_else(|e| warn!("Error registering snapshot sync protocol: {:?}", e)); - + // register the light protocol. if let Some(light_proto) = self.light_proto.as_ref().map(|x| x.clone()) { self.network.register_protocol(light_proto, self.light_subprotocol_name, ::light::net::PACKET_COUNT, ::light::net::PROTOCOL_VERSIONS) @@ -335,6 +335,11 @@ impl ChainNotify for EthSync { self.sync_handler.snapshot_service.abort_restore(); self.network.stop().unwrap_or_else(|e| warn!("Error stopping network: {:?}", e)); } + + fn transactions_received(&self, hashes: Vec, peer_id: PeerId) { + let mut sync = self.sync_handler.sync.write(); + sync.transactions_received(hashes, peer_id); + } } /// LES event handler. @@ -344,7 +349,7 @@ struct TxRelay(Arc); impl LightHandler for TxRelay { fn on_transactions(&self, ctx: &EventContext, relay: &[::ethcore::transaction::SignedTransaction]) { trace!(target: "les", "Relaying {} transactions from peer {}", relay.len(), ctx.peer()); - self.0.queue_transactions(relay.iter().map(|tx| ::rlp::encode(tx).to_vec()).collect()) + self.0.queue_transactions(relay.iter().map(|tx| ::rlp::encode(tx).to_vec()).collect(), ctx.peer()) } } @@ -547,4 +552,4 @@ pub struct ServiceConfiguration { pub net: NetworkConfiguration, /// IPC path. pub io_path: String, -} \ No newline at end of file +} diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 7b9257c6c..350a42d0e 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -432,6 +432,13 @@ impl ChainSync { self.transactions_stats.stats() } + /// Updates transactions were received by a peer + pub fn transactions_received(&mut self, hashes: Vec, peer_id: PeerId) { + if let Some(mut peer_info) = self.peers.get_mut(&peer_id) { + peer_info.last_sent_transactions.extend(&hashes); + } + } + /// Abort all sync activity pub fn abort(&mut self, io: &mut SyncIo) { self.reset_and_continue(io); @@ -1409,7 +1416,7 @@ impl ChainSync { let tx = rlp.as_raw().to_vec(); transactions.push(tx); } - io.chain().queue_transactions(transactions); + io.chain().queue_transactions(transactions, peer_id); Ok(()) } diff --git a/sync/src/transactions_stats.rs b/sync/src/transactions_stats.rs index d5b78610d..03549fb3c 100644 --- a/sync/src/transactions_stats.rs +++ b/sync/src/transactions_stats.rs @@ -112,7 +112,7 @@ mod tests { propagated_to: hash_map![ enodeid1 => 2, enodeid2 => 1 - ] + ], })); } diff --git a/util/network/src/lib.rs b/util/network/src/lib.rs index 5a06f5a22..8d85731b0 100644 --- a/util/network/src/lib.rs +++ b/util/network/src/lib.rs @@ -99,7 +99,7 @@ pub use stats::NetworkStats; pub use session::SessionInfo; use io::TimerToken; -pub use node_table::is_valid_node_url; +pub use node_table::{is_valid_node_url, NodeId}; const PROTOCOL_VERSION: u32 = 4;