title: Async read from UdpSocket
body: <p>I am trying to concurrently process arriving UDP packets in Tokio. However the following MWE does not do what I expected:</p>
  
  <pre><code>extern crate futures;
  extern crate tokio_core;
  extern crate tokio_io;
  
  use futures::{Future, Stream};
  use std::net::SocketAddr;
  use tokio_core::net::{UdpCodec, UdpSocket};
  use tokio_core::reactor::Core;
  
  // just a codec to send and receive bytes
  pub struct LineCodec;
  impl UdpCodec for LineCodec {
      type In = (SocketAddr, Vec&lt;u8&gt;);
      type Out = (SocketAddr, Vec&lt;u8&gt;);
  
      fn decode(&amp;mut self, addr: &amp;SocketAddr, buf: &amp;[u8]) -&gt; std::io::Result&lt;Self::In&gt; {
          Ok((*addr, buf.to_vec()))
      }
  
      fn encode(&amp;mut self, (addr, buf): Self::Out, into: &amp;mut Vec&lt;u8&gt;) -&gt; SocketAddr {
          into.extend(buf);
          addr
      }
  }
  
  fn compute(addr: SocketAddr, msg: Vec&lt;u8&gt;) -&gt; Box&lt;Future&lt;Item = (), Error = ()&gt;&gt; {
      println!("Starting to compute for: {}", addr);
      // sleep is a placeholder for a long computation
      std::thread::sleep(std::time::Duration::from_secs(8));
      println!("Done computing for for: {}", addr);
      Box::new(futures::future::ok(()))
  }
  
  fn main() {
      let mut core = Core::new().unwrap();
      let handle = core.handle();
      let listening_addr = "127.0.0.1:8080".parse::&lt;SocketAddr&gt;().unwrap();
      let socket = UdpSocket::bind(&amp;listening_addr, &amp;handle).unwrap();
      println!("Listening on: {}", socket.local_addr().unwrap());
  
      let (writer, reader) = socket.framed(LineCodec).split();
  
      let socket_read = reader.for_each(|(addr, msg)| {
          println!("Got {:?}", msg);
          handle.spawn(compute(addr, msg));
          Ok(())
      });
  
      core.run(socket_read).unwrap();
  }
  </code></pre>
  
  <p>Connecting two terminals with <code>$ nc -u localhost 8080</code> and sending some text, I can see that the message from the second terminal is processed after the first finished.</p>
  
  <p>What do I have to change?</p>