grassrootseconomics/openethereum
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Port try macro to new ? operator. (#3962)

Browse Source 
* initial untry sweep

* restore try in ipc codegen, fix inference

* change a few missed try instances
This commit is contained in:
Robert Habermeier
2016-12-27 12:53:56 +01:00
committed by Arkadiy Paronyan
parent b1ef52a6d7
commit 8125b5690c
165 changed files with 1696 additions and 1696 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
util/network/src/connection.rs
View File

@@ -150,7 +150,7 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
},
Ok(Some(_)) => { panic!("Wrote past buffer");},
Ok(None) => Ok(WriteStatus::Ongoing),
Err(e) => try!(Err(e))
Err(e) => Err(e)?
}
}.and_then(|r| {
if r == WriteStatus::Complete {
@@ -158,7 +158,7 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
}
if self.send_queue.is_empty() {
self.interest.remove(Ready::writable());
try!(io.update_registration(self.token));
io.update_registration(self.token)?;
}
Ok(r)
})
@@ -207,7 +207,7 @@ impl Connection {
pub fn try_clone(&self) -> io::Result<Self> {
Ok(Connection {
token: self.token,
socket: try!(self.socket.try_clone()),
socket: self.socket.try_clone()?,
rec_buf: Vec::new(),
rec_size: 0,
send_queue: self.send_queue.clone(),
@@ -300,7 +300,7 @@ pub struct EncryptedConnection {
impl EncryptedConnection {
/// Create an encrypted connection out of the handshake.
pub fn new(handshake: &mut Handshake) -> Result<EncryptedConnection, NetworkError> {
let shared = try!(crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_ephemeral));
let shared = crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_ephemeral)?;
let mut nonce_material = H512::new();
if handshake.originated {
handshake.remote_nonce.copy_to(&mut nonce_material[0..32]);
@@ -334,7 +334,7 @@ impl EncryptedConnection {
ingress_mac.update(&mac_material);
ingress_mac.update(if handshake.originated { &handshake.ack_cipher } else { &handshake.auth_cipher });
let old_connection = try!(handshake.connection.try_clone());
let old_connection = handshake.connection.try_clone()?;
let connection = ::std::mem::replace(&mut handshake.connection, old_connection);
let mut enc = EncryptedConnection {
connection: connection,
@@ -396,7 +396,7 @@ impl EncryptedConnection {
let length = ((((hdec[0] as u32) << 8) + (hdec[1] as u32)) << 8) + (hdec[2] as u32);
let header_rlp = UntrustedRlp::new(&hdec[3..6]);
let protocol_id = try!(header_rlp.val_at::<u16>(0));
let protocol_id = header_rlp.val_at::<u16>(0)?;
self.payload_len = length as usize;
self.protocol_id = protocol_id;
@@ -448,19 +448,19 @@ impl EncryptedConnection {
/// Readable IO handler. Tracker receive status and returns decoded packet if avaialable.
pub fn readable<Message>(&mut self, io: &IoContext<Message>) -> Result<Option<Packet>, NetworkError> where Message: Send + Clone + Sync + 'static {
try!(io.clear_timer(self.connection.token));
io.clear_timer(self.connection.token)?;
if let EncryptedConnectionState::Header = self.read_state {
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_header(&data));
try!(io.register_timer(self.connection.token, RECIEVE_PAYLOAD_TIMEOUT));
if let Some(data) = self.connection.readable()? {
self.read_header(&data)?;
io.register_timer(self.connection.token, RECIEVE_PAYLOAD_TIMEOUT)?;
}
};
if let EncryptedConnectionState::Payload = self.read_state {
match try!(self.connection.readable()) {
match self.connection.readable()? {
Some(data) => {
self.read_state = EncryptedConnectionState::Header;
self.connection.expect(ENCRYPTED_HEADER_LEN);
Ok(Some(try!(self.read_payload(&data))))
Ok(Some(self.read_payload(&data)?))
},
None => Ok(None)
}
@@ -471,7 +471,7 @@ impl EncryptedConnection {
/// Writable IO handler. Processes send queeue.
pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<(), NetworkError> where Message: Send + Clone + Sync + 'static {
try!(self.connection.writable(io));
self.connection.writable(io)?;
Ok(())
}
}

30
util/network/src/discovery.rs
View File

@@ -375,7 +375,7 @@ impl Discovery {
let signed = &packet[(32 + 65)..];
let signature = H520::from_slice(&packet[32..(32 + 65)]);
let node_id = try!(recover(&signature.into(), &signed.sha3()));
let node_id = recover(&signature.into(), &signed.sha3())?;
let packet_id = signed[0];
let rlp = UntrustedRlp::new(&signed[1..]);
@@ -405,10 +405,10 @@ impl Discovery {
fn on_ping(&mut self, rlp: &UntrustedRlp, node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
trace!(target: "discovery", "Got Ping from {:?}", &from);
let source = try!(NodeEndpoint::from_rlp(&try!(rlp.at(1))));
let dest = try!(NodeEndpoint::from_rlp(&try!(rlp.at(2))));
let timestamp: u64 = try!(rlp.val_at(3));
try!(self.check_timestamp(timestamp));
let source = NodeEndpoint::from_rlp(&rlp.at(1)?)?;
let dest = NodeEndpoint::from_rlp(&rlp.at(2)?)?;
let timestamp: u64 = rlp.val_at(3)?;
self.check_timestamp(timestamp)?;
let mut added_map = HashMap::new();
let entry = NodeEntry { id: node.clone(), endpoint: source.clone() };
if !entry.endpoint.is_valid() {
@@ -431,9 +431,9 @@ impl Discovery {
fn on_pong(&mut self, rlp: &UntrustedRlp, node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
trace!(target: "discovery", "Got Pong from {:?}", &from);
// TODO: validate pong packet
let dest = try!(NodeEndpoint::from_rlp(&try!(rlp.at(0))));
let timestamp: u64 = try!(rlp.val_at(2));
try!(self.check_timestamp(timestamp));
let dest = NodeEndpoint::from_rlp(&rlp.at(0)?)?;
let timestamp: u64 = rlp.val_at(2)?;
self.check_timestamp(timestamp)?;
let mut entry = NodeEntry { id: node.clone(), endpoint: dest };
if !entry.endpoint.is_valid() {
debug!(target: "discovery", "Bad address: {:?}", entry);
@@ -447,9 +447,9 @@ impl Discovery {
fn on_find_node(&mut self, rlp: &UntrustedRlp, _node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
trace!(target: "discovery", "Got FindNode from {:?}", &from);
let target: NodeId = try!(rlp.val_at(0));
let timestamp: u64 = try!(rlp.val_at(1));
try!(self.check_timestamp(timestamp));
let target: NodeId = rlp.val_at(0)?;
let timestamp: u64 = rlp.val_at(1)?;
self.check_timestamp(timestamp)?;
let nearest = Discovery::nearest_node_entries(&target, &self.node_buckets);
if nearest.is_empty() {
return Ok(None);
@@ -481,14 +481,14 @@ impl Discovery {
fn on_neighbours(&mut self, rlp: &UntrustedRlp, _node: &NodeId, from: &SocketAddr) -> Result<Option<TableUpdates>, NetworkError> {
// TODO: validate packet
let mut added = HashMap::new();
trace!(target: "discovery", "Got {} Neighbours from {:?}", try!(rlp.at(0)).item_count(), &from);
for r in try!(rlp.at(0)).iter() {
let endpoint = try!(NodeEndpoint::from_rlp(&r));
trace!(target: "discovery", "Got {} Neighbours from {:?}", rlp.at(0)?.item_count(), &from);
for r in rlp.at(0)?.iter() {
let endpoint = NodeEndpoint::from_rlp(&r)?;
if !endpoint.is_valid() {
debug!(target: "discovery", "Bad address: {:?}", endpoint);
continue;
}
let node_id: NodeId = try!(r.val_at(3));
let node_id: NodeId = r.val_at(3)?;
if node_id == self.id {
continue;
}

56
util/network/src/handshake.rs
View File

@@ -89,7 +89,7 @@ impl Handshake {
connection: Connection::new(token, socket, stats),
originated: false,
state: HandshakeState::New,
ecdhe: try!(Random.generate()),
ecdhe: Random.generate()?,
nonce: nonce.clone(),
remote_ephemeral: Public::new(),
remote_nonce: H256::new(),
@@ -110,7 +110,7 @@ impl Handshake {
self.originated = originated;
io.register_timer(self.connection.token, HANDSHAKE_TIMEOUT).ok();
if originated {
try!(self.write_auth(io, host.secret(), host.id()));
self.write_auth(io, host.secret(), host.id())?;
}
else {
self.state = HandshakeState::ReadingAuth;
@@ -127,21 +127,21 @@ impl Handshake {
/// Readable IO handler. Drives the state change.
pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<(), NetworkError> where Message: Send + Clone + Sync + 'static {
if !self.expired() {
while let Some(data) = try!(self.connection.readable()) {
while let Some(data) = self.connection.readable()? {
match self.state {
HandshakeState::New => {},
HandshakeState::StartSession => {},
HandshakeState::ReadingAuth => {
try!(self.read_auth(io, host.secret(), &data));
self.read_auth(io, host.secret(), &data)?;
},
HandshakeState::ReadingAuthEip8 => {
try!(self.read_auth_eip8(io, host.secret(), &data));
self.read_auth_eip8(io, host.secret(), &data)?;
},
HandshakeState::ReadingAck => {
try!(self.read_ack(host.secret(), &data));
self.read_ack(host.secret(), &data)?;
},
HandshakeState::ReadingAckEip8 => {
try!(self.read_ack_eip8(host.secret(), &data));
self.read_ack_eip8(host.secret(), &data)?;
},
}
if self.state == HandshakeState::StartSession {
@@ -156,7 +156,7 @@ impl Handshake {
/// Writabe IO handler.
pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<(), NetworkError> where Message: Send + Clone + Sync + 'static {
if !self.expired() {
try!(self.connection.writable(io));
self.connection.writable(io)?;
}
Ok(())
}
@@ -165,9 +165,9 @@ impl Handshake {
self.id.clone_from_slice(remote_public);
self.remote_nonce.clone_from_slice(remote_nonce);
self.remote_version = remote_version;
let shared = try!(ecdh::agree(host_secret, &self.id));
let shared = ecdh::agree(host_secret, &self.id)?;
let signature = H520::from_slice(sig);
self.remote_ephemeral = try!(recover(&signature.into(), &(&shared ^ &self.remote_nonce)));
self.remote_ephemeral = recover(&signature.into(), &(&shared ^ &self.remote_nonce))?;
Ok(())
}
@@ -185,8 +185,8 @@ impl Handshake {
let (_, rest) = rest.split_at(32);
let (pubk, rest) = rest.split_at(64);
let (nonce, _) = rest.split_at(32);
try!(self.set_auth(secret, sig, pubk, nonce, PROTOCOL_VERSION));
try!(self.write_ack(io));
self.set_auth(secret, sig, pubk, nonce, PROTOCOL_VERSION)?;
self.write_ack(io)?;
}
Err(_) => {
// Try to interpret as EIP-8 packet
@@ -206,14 +206,14 @@ impl Handshake {
fn read_auth_eip8<Message>(&mut self, io: &IoContext<Message>, secret: &Secret, data: &[u8]) -> Result<(), NetworkError> where Message: Send + Clone + Sync + 'static {
trace!(target: "network", "Received EIP8 handshake auth from {:?}", self.connection.remote_addr_str());
self.auth_cipher.extend_from_slice(data);
let auth = try!(ecies::decrypt(secret, &self.auth_cipher[0..2], &self.auth_cipher[2..]));
let auth = ecies::decrypt(secret, &self.auth_cipher[0..2], &self.auth_cipher[2..])?;
let rlp = UntrustedRlp::new(&auth);
let signature: H520 = try!(rlp.val_at(0));
let remote_public: Public = try!(rlp.val_at(1));
let remote_nonce: H256 = try!(rlp.val_at(2));
let remote_version: u64 = try!(rlp.val_at(3));
try!(self.set_auth(secret, &signature, &remote_public, &remote_nonce, remote_version));
try!(self.write_ack_eip8(io));
let signature: H520 = rlp.val_at(0)?;
let remote_public: Public = rlp.val_at(1)?;
let remote_nonce: H256 = rlp.val_at(2)?;
let remote_version: u64 = rlp.val_at(3)?;
self.set_auth(secret, &signature, &remote_public, &remote_nonce, remote_version)?;
self.write_ack_eip8(io)?;
Ok(())
}
@@ -249,11 +249,11 @@ impl Handshake {
fn read_ack_eip8(&mut self, secret: &Secret, data: &[u8]) -> Result<(), NetworkError> {
trace!(target: "network", "Received EIP8 handshake auth from {:?}", self.connection.remote_addr_str());
self.ack_cipher.extend_from_slice(data);
let ack = try!(ecies::decrypt(secret, &self.ack_cipher[0..2], &self.ack_cipher[2..]));
let ack = ecies::decrypt(secret, &self.ack_cipher[0..2], &self.ack_cipher[2..])?;
let rlp = UntrustedRlp::new(&ack);
self.remote_ephemeral = try!(rlp.val_at(0));
self.remote_nonce = try!(rlp.val_at(1));
self.remote_version = try!(rlp.val_at(2));
self.remote_ephemeral = rlp.val_at(0)?;
self.remote_nonce = rlp.val_at(1)?;
self.remote_version = rlp.val_at(2)?;
self.state = HandshakeState::StartSession;
Ok(())
}
@@ -271,13 +271,13 @@ impl Handshake {
let (nonce, _) = rest.split_at_mut(32);
// E(remote-pubk, S(ecdhe-random, ecdh-shared-secret^nonce) || H(ecdhe-random-pubk) || pubk || nonce || 0x0)
let shared = try!(ecdh::agree(secret, &self.id));
sig.copy_from_slice(&*try!(sign(self.ecdhe.secret(), &(&shared ^ &self.nonce))));
let shared = ecdh::agree(secret, &self.id)?;
sig.copy_from_slice(&*sign(self.ecdhe.secret(), &(&shared ^ &self.nonce))?);
self.ecdhe.public().sha3_into(hepubk);
pubk.copy_from_slice(public);
nonce.copy_from_slice(&self.nonce);
}
let message = try!(ecies::encrypt(&self.id, &[], &data));
let message = ecies::encrypt(&self.id, &[], &data)?;
self.auth_cipher = message.clone();
self.connection.send(io, message);
self.connection.expect(V4_ACK_PACKET_SIZE);
@@ -297,7 +297,7 @@ impl Handshake {
self.ecdhe.public().copy_to(epubk);
self.nonce.copy_to(nonce);
}
let message = try!(ecies::encrypt(&self.id, &[], &data));
let message = ecies::encrypt(&self.id, &[], &data)?;
self.ack_cipher = message.clone();
self.connection.send(io, message);
self.state = HandshakeState::StartSession;
@@ -319,7 +319,7 @@ impl Handshake {
let encoded = rlp.drain();
let len = (encoded.len() + ECIES_OVERHEAD) as u16;
let prefix = [ (len >> 8) as u8, (len & 0xff) as u8 ];
let message = try!(ecies::encrypt(&self.id, &prefix, &encoded));
let message = ecies::encrypt(&self.id, &prefix, &encoded)?;
self.ack_cipher.extend_from_slice(&prefix);
self.ack_cipher.extend_from_slice(&message);
self.connection.send(io, self.ack_cipher.clone());

26
util/network/src/host.rs
View File

@@ -251,7 +251,7 @@ impl<'s> NetworkContext<'s> {
pub fn send_protocol(&self, protocol: ProtocolId, peer: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
let session = self.resolve_session(peer);
if let Some(session) = session {
try!(session.lock().send_packet(self.io, protocol, packet_id as u8, &data));
session.lock().send_packet(self.io, protocol, packet_id as u8, &data)?;
} else {
trace!(target: "network", "Send: Peer no longer exist")
}
@@ -387,7 +387,7 @@ impl Host {
};
let keys = if let Some(ref secret) = config.use_secret {
try!(KeyPair::from_secret(secret.clone()))
KeyPair::from_secret(secret.clone())?
} else {
config.config_path.clone().and_then(|ref p| load_key(Path::new(&p)))
.map_or_else(|| {
@@ -401,8 +401,8 @@ impl Host {
};
let path = config.net_config_path.clone();
// Setup the server socket
let tcp_listener = try!(TcpListener::bind(&listen_address));
listen_address = SocketAddr::new(listen_address.ip(), try!(tcp_listener.local_addr()).port());
let tcp_listener = TcpListener::bind(&listen_address)?;
listen_address = SocketAddr::new(listen_address.ip(), tcp_listener.local_addr()?.port());
debug!(target: "network", "Listening at {:?}", listen_address);
let udp_port = config.udp_port.unwrap_or(listen_address.port());
let local_endpoint = NodeEndpoint { address: listen_address, udp_port: udp_port };
@@ -462,7 +462,7 @@ impl Host {
}
pub fn add_reserved_node(&self, id: &str) -> Result<(), NetworkError> {
let n = try!(Node::from_str(id));
let n = Node::from_str(id)?;
let entry = NodeEntry { endpoint: n.endpoint.clone(), id: n.id.clone() };
self.reserved_nodes.write().insert(n.id.clone());
@@ -506,7 +506,7 @@ impl Host {
}
pub fn remove_reserved_node(&self, id: &str) -> Result<(), NetworkError> {
let n = try!(Node::from_str(id));
let n = Node::from_str(id)?;
self.reserved_nodes.write().remove(&n.id);
Ok(())
@@ -538,7 +538,7 @@ impl Host {
trace!(target: "network", "Disconnecting on shutdown: {}", p);
self.kill_connection(p, io, true);
}
try!(io.unregister_handler());
io.unregister_handler()?;
Ok(())
}
@@ -588,12 +588,12 @@ impl Host {
discovery.init_node_list(self.nodes.read().unordered_entries());
discovery.add_node_list(self.nodes.read().unordered_entries());
*self.discovery.lock() = Some(discovery);
try!(io.register_stream(DISCOVERY));
try!(io.register_timer(DISCOVERY_REFRESH, DISCOVERY_REFRESH_TIMEOUT));
try!(io.register_timer(DISCOVERY_ROUND, DISCOVERY_ROUND_TIMEOUT));
io.register_stream(DISCOVERY)?;
io.register_timer(DISCOVERY_REFRESH, DISCOVERY_REFRESH_TIMEOUT)?;
io.register_timer(DISCOVERY_ROUND, DISCOVERY_ROUND_TIMEOUT)?;
}
try!(io.register_timer(NODE_TABLE, NODE_TABLE_TIMEOUT));
try!(io.register_stream(TCP_ACCEPT));
io.register_timer(NODE_TABLE, NODE_TABLE_TIMEOUT)?;
io.register_stream(TCP_ACCEPT)?;
Ok(())
}
@@ -737,7 +737,7 @@ impl Host {
});
match token {
Some(t) => Ok(try!(From::from(io.register_stream(t)))),
Some(t) => io.register_stream(t).map(|_| ()).map_err(Into::into),
None => {
debug!(target: "network", "Max sessions reached");
Ok(())

18
util/network/src/node_table.rs
View File

@@ -64,10 +64,10 @@ impl NodeEndpoint {
}
pub fn from_rlp(rlp: &UntrustedRlp) -> Result<Self, DecoderError> {
let tcp_port = try!(rlp.val_at::<u16>(2));
let udp_port = try!(rlp.val_at::<u16>(1));
let addr_bytes = try!(try!(rlp.at(0)).data());
let address = try!(match addr_bytes.len() {
let tcp_port = rlp.val_at::<u16>(2)?;
let udp_port = rlp.val_at::<u16>(1)?;
let addr_bytes = rlp.at(0)?.data()?;
let address = match addr_bytes.len() {
4 => Ok(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(addr_bytes[0], addr_bytes[1], addr_bytes[2], addr_bytes[3]), tcp_port))),
16 => unsafe {
let o: *const u16 = mem::transmute(addr_bytes.as_ptr());
@@ -75,7 +75,7 @@ impl NodeEndpoint {
Ok(SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::new(o[0], o[1], o[2], o[3], o[4], o[5], o[6], o[7]), tcp_port, 0, 0)))
},
_ => Err(DecoderError::RlpInconsistentLengthAndData)
});
}?;
Ok(NodeEndpoint { address: address, udp_port: udp_port })
}
@@ -153,9 +153,9 @@ impl Node {
impl Display for Node {
fn fmt(&self, f: &mut Formatter) -> fmt::Result {
if self.endpoint.udp_port != self.endpoint.address.port() {
try!(write!(f, "enode://{}@{}+{}", self.id.hex(), self.endpoint.address, self.endpoint.udp_port));
write!(f, "enode://{}@{}+{}", self.id.hex(), self.endpoint.address, self.endpoint.udp_port)?;
} else {
try!(write!(f, "enode://{}@{}", self.id.hex(), self.endpoint.address));
write!(f, "enode://{}@{}", self.id.hex(), self.endpoint.address)?;
}
Ok(())
}
@@ -165,10 +165,10 @@ impl FromStr for Node {
type Err = NetworkError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let (id, endpoint) = if s.len() > 136 && &s[0..8] == "enode://" && &s[136..137] == "@" {
(try!(s[8..136].parse().map_err(UtilError::from)), try!(NodeEndpoint::from_str(&s[137..])))
(s[8..136].parse().map_err(UtilError::from)?, NodeEndpoint::from_str(&s[137..])?)
}
else {
(NodeId::new(), try!(NodeEndpoint::from_str(s)))
(NodeId::new(), NodeEndpoint::from_str(s)?)
};
Ok(Node {

14
util/network/src/service.rs
View File

@@ -56,7 +56,7 @@ impl NetworkService {
pub fn new(config: NetworkConfiguration) -> Result<NetworkService, NetworkError> {
let host_handler = Arc::new(HostHandler { public_url: RwLock::new(None) });
let panic_handler = PanicHandler::new_in_arc();
let io_service = try!(IoService::<NetworkIoMessage>::start());
let io_service = IoService::<NetworkIoMessage>::start()?;
panic_handler.forward_from(&io_service);
let stats = Arc::new(NetworkStats::new());
@@ -74,12 +74,12 @@ impl NetworkService {
/// Regiter a new protocol handler with the event loop.
pub fn register_protocol(&self, handler: Arc<NetworkProtocolHandler + Send + Sync>, protocol: ProtocolId, packet_count: u8, versions: &[u8]) -> Result<(), NetworkError> {
try!(self.io_service.send_message(NetworkIoMessage::AddHandler {
self.io_service.send_message(NetworkIoMessage::AddHandler {
handler: handler,
protocol: protocol,
versions: versions.to_vec(),
packet_count: packet_count,
}));
})?;
Ok(())
}
@@ -119,13 +119,13 @@ impl NetworkService {
pub fn start(&self) -> Result<(), NetworkError> {
let mut host = self.host.write();
if host.is_none() {
let h = Arc::new(try!(Host::new(self.config.clone(), self.stats.clone())));
try!(self.io_service.register_handler(h.clone()));
let h = Arc::new(Host::new(self.config.clone(), self.stats.clone())?);
self.io_service.register_handler(h.clone())?;
*host = Some(h);
}
if self.host_handler.public_url.read().is_none() {
try!(self.io_service.register_handler(self.host_handler.clone()));
self.io_service.register_handler(self.host_handler.clone())?;
}
Ok(())
@@ -136,7 +136,7 @@ impl NetworkService {
let mut host = self.host.write();
if let Some(ref host) = *host {
let io = IoContext::new(self.io_service.channel(), 0); //TODO: take token id from host
try!(host.stop(&io));
host.stop(&io)?;
}
*host = None;
Ok(())

46
util/network/src/session.rs
View File

@@ -118,7 +118,7 @@ pub struct PeerCapabilityInfo {
impl Decodable for PeerCapabilityInfo {
fn decode<D>(decoder: &D) -> Result<Self, DecoderError> where D: Decoder {
let c = decoder.as_rlp();
let p: Vec<u8> = try!(c.val_at(0));
let p: Vec<u8> = c.val_at(0)?;
if p.len() != 3 {
return Err(DecoderError::Custom("Invalid subprotocol string length. Should be 3"));
}
@@ -126,7 +126,7 @@ impl Decodable for PeerCapabilityInfo {
p2.clone_from_slice(&p);
Ok(PeerCapabilityInfo {
protocol: p2,
version: try!(c.val_at(1))
version: c.val_at(1)?
})
}
}
@@ -180,7 +180,7 @@ impl Session {
let originated = id.is_some();
let mut handshake = Handshake::new(token, id, socket, nonce, stats).expect("Can't create handshake");
let local_addr = handshake.connection.local_addr_str();
try!(handshake.start(io, host, originated));
handshake.start(io, host, originated)?;
Ok(Session {
state: State::Handshake(handshake),
had_hello: false,
@@ -206,13 +206,13 @@ impl Session {
let connection = if let State::Handshake(ref mut h) = self.state {
self.info.id = Some(h.id.clone());
self.info.remote_address = h.connection.remote_addr_str();
try!(EncryptedConnection::new(h))
EncryptedConnection::new(h)?
} else {
panic!("Unexpected state");
};
self.state = State::Session(connection);
try!(self.write_hello(io, host));
try!(self.send_ping(io));
self.write_hello(io, host)?;
self.send_ping(io)?;
Ok(())
}
@@ -265,23 +265,23 @@ impl Session {
let mut packet_data = None;
match self.state {
State::Handshake(ref mut h) => {
try!(h.readable(io, host));
h.readable(io, host)?;
if h.done() {
create_session = true;
}
}
State::Session(ref mut c) => {
match try!(c.readable(io)) {
match c.readable(io)? {
data @ Some(_) => packet_data = data,
None => return Ok(SessionData::None)
}
}
}
if let Some(data) = packet_data {
return Ok(try!(self.read_packet(io, data, host)));
return Ok(self.read_packet(io, data, host)?);
}
if create_session {
try!(self.complete_handshake(io, host));
self.complete_handshake(io, host)?;
io.update_registration(self.token()).unwrap_or_else(|e| debug!(target: "network", "Token registration error: {:?}", e));
}
Ok(SessionData::None)
@@ -310,19 +310,19 @@ impl Session {
if self.expired() {
return Ok(());
}
try!(self.connection().register_socket(reg, event_loop));
self.connection().register_socket(reg, event_loop)?;
Ok(())
}
/// Update registration with the event loop. Should be called at the end of the IO handler.
pub fn update_socket<Host:Handler>(&self, reg:Token, event_loop: &mut EventLoop<Host>) -> Result<(), NetworkError> {
try!(self.connection().update_socket(reg, event_loop));
self.connection().update_socket(reg, event_loop)?;
Ok(())
}
/// Delete registration
pub fn deregister_socket<Host:Handler>(&self, event_loop: &mut EventLoop<Host>) -> Result<(), NetworkError> {
try!(self.connection().deregister_socket(event_loop));
self.connection().deregister_socket(event_loop)?;
Ok(())
}
@@ -400,19 +400,19 @@ impl Session {
match packet_id {
PACKET_HELLO => {
let rlp = UntrustedRlp::new(&packet.data[1..]); //TODO: validate rlp expected size
try!(self.read_hello(io, &rlp, host));
self.read_hello(io, &rlp, host)?;
Ok(SessionData::Ready)
},
PACKET_DISCONNECT => {
let rlp = UntrustedRlp::new(&packet.data[1..]);
let reason: u8 = try!(rlp.val_at(0));
let reason: u8 = rlp.val_at(0)?;
if self.had_hello {
debug!("Disconnected: {}: {:?}", self.token(), DisconnectReason::from_u8(reason));
}
Err(From::from(NetworkError::Disconnect(DisconnectReason::from_u8(reason))))
}
PACKET_PING => {
try!(self.send_pong(io));
self.send_pong(io)?;
Ok(SessionData::Continue)
},
PACKET_PONG => {
@@ -471,10 +471,10 @@ impl Session {
fn read_hello<Message>(&mut self, io: &IoContext<Message>, rlp: &UntrustedRlp, host: &HostInfo) -> Result<(), NetworkError>
where Message: Send + Sync + Clone {
let protocol = try!(rlp.val_at::<u32>(0));
let client_version = try!(rlp.val_at::<String>(1));
let peer_caps = try!(rlp.val_at::<Vec<PeerCapabilityInfo>>(2));
let id = try!(rlp.val_at::<NodeId>(4));
let protocol = rlp.val_at::<u32>(0)?;
let client_version = rlp.val_at::<String>(1)?;
let peer_caps = rlp.val_at::<Vec<PeerCapabilityInfo>>(2)?;
let id = rlp.val_at::<NodeId>(4)?;
// Intersect with host capabilities
// Leave only highset mutually supported capability version
@@ -530,14 +530,14 @@ impl Session {
/// Senf ping packet
pub fn send_ping<Message>(&mut self, io: &IoContext<Message>) -> Result<(), NetworkError> where Message: Send + Sync + Clone {
try!(self.send(io, try!(Session::prepare(PACKET_PING))));
self.send(io, Session::prepare(PACKET_PING)?)?;
self.ping_time_ns = time::precise_time_ns();
self.pong_time_ns = None;
Ok(())
}
fn send_pong<Message>(&mut self, io: &IoContext<Message>) -> Result<(), NetworkError> where Message: Send + Sync + Clone {
self.send(io, try!(Session::prepare(PACKET_PONG)))
self.send(io, Session::prepare(PACKET_PONG)?)
}
/// Disconnect this session
@@ -565,7 +565,7 @@ impl Session {
warn!(target:"network", "Unexpected send request");
},
State::Session(ref mut s) => {
try!(s.send_packet(io, &rlp.out()))
s.send_packet(io, &rlp.out())?
},
}
Ok(())