xmpp_codec, client: handle StreamEnd
This commit is contained in:
parent
380bd2fc02
commit
599e3be32e
3 changed files with 38 additions and 3 deletions
|
@ -164,10 +164,24 @@ impl Stream for Client {
|
||||||
Ok(Async::Ready(Some(Event::Disconnected)))
|
Ok(Async::Ready(Some(Event::Disconnected)))
|
||||||
}
|
}
|
||||||
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
|
Ok(Async::Ready(Some(Packet::Stanza(stanza)))) => {
|
||||||
|
// Receive stanza
|
||||||
self.state = ClientState::Connected(stream);
|
self.state = ClientState::Connected(stream);
|
||||||
Ok(Async::Ready(Some(Event::Stanza(stanza))))
|
Ok(Async::Ready(Some(Event::Stanza(stanza))))
|
||||||
}
|
}
|
||||||
Ok(Async::NotReady) | Ok(Async::Ready(_)) => {
|
Ok(Async::Ready(Some(Packet::Text(_)))) => {
|
||||||
|
// Ignore text between stanzas
|
||||||
|
Ok(Async::NotReady)
|
||||||
|
}
|
||||||
|
Ok(Async::Ready(Some(Packet::StreamStart(_)))) => {
|
||||||
|
// <stream:stream>
|
||||||
|
Err(ProtocolError::InvalidStreamStart.into())
|
||||||
|
}
|
||||||
|
Ok(Async::Ready(Some(Packet::StreamEnd))) => {
|
||||||
|
// End of stream: </stream:stream>
|
||||||
|
Ok(Async::Ready(None))
|
||||||
|
}
|
||||||
|
Ok(Async::NotReady) => {
|
||||||
|
// Try again later
|
||||||
self.state = ClientState::Connected(stream);
|
self.state = ClientState::Connected(stream);
|
||||||
Ok(Async::NotReady)
|
Ok(Async::NotReady)
|
||||||
}
|
}
|
||||||
|
@ -212,8 +226,8 @@ impl Sink for Client {
|
||||||
/// This closes the inner TCP stream.
|
/// This closes the inner TCP stream.
|
||||||
///
|
///
|
||||||
/// To synchronize your shutdown with the server side, you should
|
/// To synchronize your shutdown with the server side, you should
|
||||||
/// first send `Packet::StreamEnd` and wait it to be sent back
|
/// first send `Packet::StreamEnd` and wait for the end of the
|
||||||
/// before closing the connection.
|
/// incoming stream before closing the connection.
|
||||||
fn close(&mut self) -> Poll<(), Self::SinkError> {
|
fn close(&mut self) -> Poll<(), Self::SinkError> {
|
||||||
match self.state {
|
match self.state {
|
||||||
ClientState::Connected(ref mut stream) =>
|
ClientState::Connected(ref mut stream) =>
|
||||||
|
|
|
@ -88,6 +88,8 @@ pub enum ProtocolError {
|
||||||
NoStreamId,
|
NoStreamId,
|
||||||
/// Encountered an unexpected XML token
|
/// Encountered an unexpected XML token
|
||||||
InvalidToken,
|
InvalidToken,
|
||||||
|
/// Unexpected <stream:stream> (shouldn't occur)
|
||||||
|
InvalidStreamStart,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Authentication error
|
/// Authentication error
|
||||||
|
|
|
@ -378,6 +378,25 @@ mod tests {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_stream_end() {
|
||||||
|
let mut c = XMPPCodec::new();
|
||||||
|
let mut b = BytesMut::with_capacity(1024);
|
||||||
|
b.put(r"<?xml version='1.0'?><stream:stream xmlns:stream='http://etherx.jabber.org/streams' version='1.0' xmlns='jabber:client'>");
|
||||||
|
let r = c.decode(&mut b);
|
||||||
|
assert!(match r {
|
||||||
|
Ok(Some(Packet::StreamStart(_))) => true,
|
||||||
|
_ => false,
|
||||||
|
});
|
||||||
|
b.clear();
|
||||||
|
b.put(r"</stream:stream>");
|
||||||
|
let r = c.decode(&mut b);
|
||||||
|
assert!(match r {
|
||||||
|
Ok(Some(Packet::StreamEnd)) => true,
|
||||||
|
_ => false,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_truncated_stanza() {
|
fn test_truncated_stanza() {
|
||||||
let mut c = XMPPCodec::new();
|
let mut c = XMPPCodec::new();
|
||||||
|
|
Loading…
Reference in a new issue