rust/src/libstd/flatpipes.rs

971 lines
28 KiB
Rust
Raw Normal View History

2013-01-18 01:28:42 -06:00
// Copyright 2013 The Rust Project Developers. See the COPYRIGHT
// file at the top-level directory of this distribution and at
// http://rust-lang.org/COPYRIGHT.
//
// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. This file may not be copied, modified, or distributed
// except according to those terms.
2012-11-25 16:12:21 -06:00
/*!
Generic communication channels for things that can be represented as,
or transformed to and from, byte vectors.
The `FlatPort` and `FlatChan` types implement the generic channel and
port interface for arbitrary types and transport strategies. It can
particularly be used to send and recieve serializable types over I/O
streams.
`FlatPort` and `FlatChan` implement the same comm traits as pipe-based
ports and channels.
# Example
This example sends boxed integers across tasks using serialization.
~~~
let (port, chan) = serial::pipe_stream();
2013-02-15 01:30:30 -06:00
do task::spawn || {
2012-11-25 16:12:21 -06:00
for int::range(0, 10) |i| {
chan.send(@i)
}
}
for int::range(0, 10) |i| {
assert @i == port.recv()
}
~~~
# Safety Note
Flat pipes created from `io::Reader`s and `io::Writer`s share the same
blocking properties as the underlying stream. Since some implementations
block the scheduler thread, so will their pipes.
*/
// The basic send/recv interface FlatChan and PortChan will implement
use core::io;
2013-02-02 05:10:12 -06:00
use core::comm::GenericChan;
use core::comm::GenericPort;
use core::prelude::*;
2012-11-25 16:12:21 -06:00
use core::sys::size_of;
use core::vec;
2012-11-25 16:12:21 -06:00
/**
A FlatPort, consisting of a `BytePort` that recieves byte vectors,
and an `Unflattener` that converts the bytes to a value.
Create using the constructors in the `serial` and `pod` modules.
*/
pub struct FlatPort<T, U, P> {
2012-11-25 16:12:21 -06:00
unflattener: U,
byte_port: P
}
/**
A FlatChan, consisting of a `Flattener` that converts values to
byte vectors, and a `ByteChan` that transmits the bytes.
Create using the constructors in the `serial` and `pod` modules.
*/
pub struct FlatChan<T, F, C> {
2012-11-25 16:12:21 -06:00
flattener: F,
byte_chan: C
}
/**
Constructors for flat pipes that using serialization-based flattening.
*/
pub mod serial {
2012-12-17 22:23:09 -06:00
pub use DefaultEncoder = ebml::writer::Encoder;
pub use DefaultDecoder = ebml::reader::Decoder;
2012-11-25 16:12:21 -06:00
2012-12-17 22:23:09 -06:00
use serialize::{Decodable, Encodable};
2012-11-25 16:12:21 -06:00
use flatpipes::flatteners::{DeserializingUnflattener,
SerializingFlattener};
use flatpipes::flatteners::{deserialize_buffer, serialize_value};
use flatpipes::bytepipes::{ReaderBytePort, WriterByteChan};
use flatpipes::bytepipes::{PipeBytePort, PipeByteChan};
use flatpipes::{FlatPort, FlatChan};
use core::io::{Reader, Writer};
2013-02-02 05:10:12 -06:00
use core::comm::{Port, Chan};
use core::comm;
2012-11-25 16:12:21 -06:00
pub type ReaderPort<T, R> = FlatPort<
2012-12-17 22:23:09 -06:00
T, DeserializingUnflattener<DefaultDecoder, T>,
2012-11-25 16:12:21 -06:00
ReaderBytePort<R>>;
pub type WriterChan<T, W> = FlatChan<
2012-12-17 22:23:09 -06:00
T, SerializingFlattener<DefaultEncoder, T>, WriterByteChan<W>>;
2012-11-25 16:12:21 -06:00
pub type PipePort<T> = FlatPort<
2012-12-17 22:23:09 -06:00
T, DeserializingUnflattener<DefaultDecoder, T>, PipeBytePort>;
2012-11-25 16:12:21 -06:00
pub type PipeChan<T> = FlatChan<
2012-12-17 22:23:09 -06:00
T, SerializingFlattener<DefaultEncoder, T>, PipeByteChan>;
2012-11-25 16:12:21 -06:00
/// Create a `FlatPort` from a `Reader`
2012-12-17 22:23:09 -06:00
pub fn reader_port<T: Decodable<DefaultDecoder>,
2012-11-25 16:12:21 -06:00
R: Reader>(reader: R) -> ReaderPort<T, R> {
2012-12-17 22:23:09 -06:00
let unflat: DeserializingUnflattener<DefaultDecoder, T> =
2012-11-25 16:12:21 -06:00
DeserializingUnflattener::new(
2012-12-17 22:23:09 -06:00
deserialize_buffer::<DefaultDecoder, T>);
2013-02-15 01:30:30 -06:00
let byte_port = ReaderBytePort::new(reader);
FlatPort::new(unflat, byte_port)
2012-11-25 16:12:21 -06:00
}
/// Create a `FlatChan` from a `Writer`
2012-12-17 22:23:09 -06:00
pub fn writer_chan<T: Encodable<DefaultEncoder>,
2012-11-25 16:12:21 -06:00
W: Writer>(writer: W) -> WriterChan<T, W> {
2012-12-17 22:23:09 -06:00
let flat: SerializingFlattener<DefaultEncoder, T> =
2012-11-25 16:12:21 -06:00
SerializingFlattener::new(
2012-12-17 22:23:09 -06:00
serialize_value::<DefaultEncoder, T>);
2013-02-15 01:30:30 -06:00
let byte_chan = WriterByteChan::new(writer);
FlatChan::new(flat, byte_chan)
2012-11-25 16:12:21 -06:00
}
/// Create a `FlatPort` from a `Port<~[u8]>`
pub fn pipe_port<T:Decodable<DefaultDecoder>>(
2012-11-25 16:12:21 -06:00
port: Port<~[u8]>
) -> PipePort<T> {
2012-12-17 22:23:09 -06:00
let unflat: DeserializingUnflattener<DefaultDecoder, T> =
2012-11-25 16:12:21 -06:00
DeserializingUnflattener::new(
2012-12-17 22:23:09 -06:00
deserialize_buffer::<DefaultDecoder, T>);
2013-02-15 01:30:30 -06:00
let byte_port = PipeBytePort::new(port);
FlatPort::new(unflat, byte_port)
2012-11-25 16:12:21 -06:00
}
/// Create a `FlatChan` from a `Chan<~[u8]>`
pub fn pipe_chan<T:Encodable<DefaultEncoder>>(
2012-11-25 16:12:21 -06:00
chan: Chan<~[u8]>
) -> PipeChan<T> {
2012-12-17 22:23:09 -06:00
let flat: SerializingFlattener<DefaultEncoder, T> =
2012-11-25 16:12:21 -06:00
SerializingFlattener::new(
2012-12-17 22:23:09 -06:00
serialize_value::<DefaultEncoder, T>);
2013-02-15 01:30:30 -06:00
let byte_chan = PipeByteChan::new(chan);
FlatChan::new(flat, byte_chan)
2012-11-25 16:12:21 -06:00
}
/// Create a pair of `FlatChan` and `FlatPort`, backed by pipes
pub fn pipe_stream<T: Encodable<DefaultEncoder> +
2012-12-17 22:23:09 -06:00
Decodable<DefaultDecoder>>(
2012-11-25 16:12:21 -06:00
) -> (PipePort<T>, PipeChan<T>) {
2013-02-02 05:10:12 -06:00
let (port, chan) = comm::stream();
2013-02-15 01:30:30 -06:00
return (pipe_port(port), pipe_chan(chan));
2012-11-25 16:12:21 -06:00
}
}
// FIXME #4074 this doesn't correctly enforce POD bounds
/**
Constructors for flat pipes that send POD types using memcpy.
# Safety Note
This module is currently unsafe because it uses `Copy Owned` as a type
parameter bounds meaning POD (plain old data), but `Copy Owned` and
POD are not equivelant.
*/
pub mod pod {
use flatpipes::flatteners::{PodUnflattener, PodFlattener};
use flatpipes::bytepipes::{ReaderBytePort, WriterByteChan};
use flatpipes::bytepipes::{PipeBytePort, PipeByteChan};
use flatpipes::{FlatPort, FlatChan};
use core::io::{Reader, Writer};
2013-02-02 05:10:12 -06:00
use core::comm::{Port, Chan};
use core::comm;
use core::prelude::*;
pub type ReaderPort<T, R> =
2012-11-25 16:12:21 -06:00
FlatPort<T, PodUnflattener<T>, ReaderBytePort<R>>;
pub type WriterChan<T, W> =
2012-11-25 16:12:21 -06:00
FlatChan<T, PodFlattener<T>, WriterByteChan<W>>;
pub type PipePort<T> = FlatPort<T, PodUnflattener<T>, PipeBytePort>;
pub type PipeChan<T> = FlatChan<T, PodFlattener<T>, PipeByteChan>;
2012-11-25 16:12:21 -06:00
/// Create a `FlatPort` from a `Reader`
pub fn reader_port<T:Copy + Owned,R:Reader>(
2012-11-25 16:12:21 -06:00
reader: R
) -> ReaderPort<T, R> {
let unflat: PodUnflattener<T> = PodUnflattener::new();
2013-02-15 01:30:30 -06:00
let byte_port = ReaderBytePort::new(reader);
FlatPort::new(unflat, byte_port)
2012-11-25 16:12:21 -06:00
}
/// Create a `FlatChan` from a `Writer`
pub fn writer_chan<T:Copy + Owned,W:Writer>(
2012-11-25 16:12:21 -06:00
writer: W
) -> WriterChan<T, W> {
let flat: PodFlattener<T> = PodFlattener::new();
2013-02-15 01:30:30 -06:00
let byte_chan = WriterByteChan::new(writer);
FlatChan::new(flat, byte_chan)
2012-11-25 16:12:21 -06:00
}
/// Create a `FlatPort` from a `Port<~[u8]>`
pub fn pipe_port<T:Copy + Owned>(port: Port<~[u8]>) -> PipePort<T> {
2012-11-25 16:12:21 -06:00
let unflat: PodUnflattener<T> = PodUnflattener::new();
2013-02-15 01:30:30 -06:00
let byte_port = PipeBytePort::new(port);
FlatPort::new(unflat, byte_port)
2012-11-25 16:12:21 -06:00
}
/// Create a `FlatChan` from a `Chan<~[u8]>`
pub fn pipe_chan<T:Copy + Owned>(chan: Chan<~[u8]>) -> PipeChan<T> {
2012-11-25 16:12:21 -06:00
let flat: PodFlattener<T> = PodFlattener::new();
2013-02-15 01:30:30 -06:00
let byte_chan = PipeByteChan::new(chan);
FlatChan::new(flat, byte_chan)
2012-11-25 16:12:21 -06:00
}
/// Create a pair of `FlatChan` and `FlatPort`, backed by pipes
pub fn pipe_stream<T:Copy + Owned>() -> (PipePort<T>, PipeChan<T>) {
2013-02-02 05:10:12 -06:00
let (port, chan) = comm::stream();
2013-02-15 01:30:30 -06:00
return (pipe_port(port), pipe_chan(chan));
2012-11-25 16:12:21 -06:00
}
}
/**
Flatteners present a value as a byte vector
*/
pub trait Flattener<T> {
fn flatten(&self, val: T) -> ~[u8];
}
/**
Unflatteners convert a byte vector to a value
*/
pub trait Unflattener<T> {
fn unflatten(&self, buf: ~[u8]) -> T;
}
/**
BytePorts are a simple interface for receiving a specified number
*/
pub trait BytePort {
fn try_recv(&self, count: uint) -> Option<~[u8]>;
}
/**
ByteChans are a simple interface for sending bytes
*/
pub trait ByteChan {
fn send(&self, val: ~[u8]);
}
static CONTINUE: [u8, ..4] = [0xAA, 0xBB, 0xCC, 0xDD];
2012-11-25 16:12:21 -06:00
impl<T,U:Unflattener<T>,P:BytePort> GenericPort<T> for FlatPort<T, U, P> {
2013-03-04 21:36:15 -06:00
fn recv(&self) -> T {
2012-11-25 16:12:21 -06:00
match self.try_recv() {
2013-02-15 01:30:30 -06:00
Some(val) => val,
None => fail!(~"port is closed")
2012-11-25 16:12:21 -06:00
}
}
2013-03-04 21:36:15 -06:00
fn try_recv(&self) -> Option<T> {
2012-11-25 16:12:21 -06:00
let command = match self.byte_port.try_recv(CONTINUE.len()) {
2013-02-15 01:30:30 -06:00
Some(c) => c,
2012-11-25 16:12:21 -06:00
None => {
warn!("flatpipe: broken pipe");
return None;
}
};
if vec::eq(command, CONTINUE) {
let msg_len = match self.byte_port.try_recv(size_of::<u64>()) {
Some(bytes) => {
io::u64_from_be_bytes(bytes, 0, size_of::<u64>())
},
None => {
warn!("flatpipe: broken pipe");
return None;
}
};
let msg_len = msg_len as uint;
match self.byte_port.try_recv(msg_len) {
2013-02-15 01:30:30 -06:00
Some(bytes) => {
Some(self.unflattener.unflatten(bytes))
2012-11-25 16:12:21 -06:00
}
None => {
warn!("flatpipe: broken pipe");
return None;
}
}
}
else {
fail!(~"flatpipe: unrecognized command");
2012-11-25 16:12:21 -06:00
}
}
}
impl<T,F:Flattener<T>,C:ByteChan> GenericChan<T> for FlatChan<T, F, C> {
2013-03-04 21:36:15 -06:00
fn send(&self, val: T) {
2012-11-25 16:12:21 -06:00
self.byte_chan.send(CONTINUE.to_vec());
2013-02-15 01:30:30 -06:00
let bytes = self.flattener.flatten(val);
2012-11-25 16:12:21 -06:00
let len = bytes.len() as u64;
do io::u64_to_be_bytes(len, size_of::<u64>()) |len_bytes| {
self.byte_chan.send(len_bytes.to_vec());
}
2013-02-15 01:30:30 -06:00
self.byte_chan.send(bytes);
2012-11-25 16:12:21 -06:00
}
}
pub impl<T,U:Unflattener<T>,P:BytePort> FlatPort<T, U, P> {
fn new(u: U, p: P) -> FlatPort<T, U, P> {
2012-11-25 16:12:21 -06:00
FlatPort {
2013-02-15 01:30:30 -06:00
unflattener: u,
byte_port: p
2012-11-25 16:12:21 -06:00
}
}
}
pub impl<T,F:Flattener<T>,C:ByteChan> FlatChan<T, F, C> {
fn new(f: F, c: C) -> FlatChan<T, F, C> {
2012-11-25 16:12:21 -06:00
FlatChan {
2013-02-15 01:30:30 -06:00
flattener: f,
byte_chan: c
2012-11-25 16:12:21 -06:00
}
}
}
pub mod flatteners {
use ebml;
2013-03-26 15:38:07 -05:00
use flatpipes::{Flattener, Unflattener};
2013-01-10 22:09:26 -06:00
use io_util::BufReader;
use json;
use serialize::{Encoder, Decoder, Encodable, Decodable};
2012-11-25 16:12:21 -06:00
use core::cast;
2013-03-26 15:38:07 -05:00
use core::io::{Writer, Reader, ReaderUtil};
use core::prelude::*;
use core::ptr;
use core::sys::size_of;
use core::vec;
2012-11-25 16:12:21 -06:00
2013-01-10 22:09:26 -06:00
// FIXME #4074: Copy + Owned != POD
pub struct PodUnflattener<T> {
2012-11-25 16:12:21 -06:00
bogus: ()
}
pub struct PodFlattener<T> {
2012-11-25 16:12:21 -06:00
bogus: ()
}
impl<T:Copy + Owned> Unflattener<T> for PodUnflattener<T> {
2012-11-25 16:12:21 -06:00
fn unflatten(&self, buf: ~[u8]) -> T {
fail_unless!(size_of::<T>() != 0);
fail_unless!(size_of::<T>() == buf.len());
2012-11-25 16:12:21 -06:00
let addr_of_init: &u8 = unsafe { &*vec::raw::to_ptr(buf) };
let addr_of_value: &T = unsafe { cast::transmute(addr_of_init) };
copy *addr_of_value
}
}
impl<T:Copy + Owned> Flattener<T> for PodFlattener<T> {
2012-11-25 16:12:21 -06:00
fn flatten(&self, val: T) -> ~[u8] {
fail_unless!(size_of::<T>() != 0);
2012-11-25 16:12:21 -06:00
let val: *T = ptr::to_unsafe_ptr(&val);
let byte_value = val as *u8;
unsafe { vec::from_buf(byte_value, size_of::<T>()) }
}
}
pub impl<T:Copy + Owned> PodUnflattener<T> {
fn new() -> PodUnflattener<T> {
2012-11-25 16:12:21 -06:00
PodUnflattener {
bogus: ()
}
}
}
pub impl<T:Copy + Owned> PodFlattener<T> {
fn new() -> PodFlattener<T> {
2012-11-25 16:12:21 -06:00
PodFlattener {
bogus: ()
}
}
}
pub type DeserializeBuffer<T> = ~fn(buf: &[u8]) -> T;
pub struct DeserializingUnflattener<D, T> {
2012-11-25 16:12:21 -06:00
deserialize_buffer: DeserializeBuffer<T>
}
pub type SerializeValue<T> = ~fn(val: &T) -> ~[u8];
pub struct SerializingFlattener<S, T> {
2012-11-25 16:12:21 -06:00
serialize_value: SerializeValue<T>
}
impl<D:Decoder,T:Decodable<D>> Unflattener<T>
for DeserializingUnflattener<D, T> {
2012-11-25 16:12:21 -06:00
fn unflatten(&self, buf: ~[u8]) -> T {
(self.deserialize_buffer)(buf)
}
}
impl<S:Encoder,T:Encodable<S>> Flattener<T>
for SerializingFlattener<S, T> {
2012-11-25 16:12:21 -06:00
fn flatten(&self, val: T) -> ~[u8] {
(self.serialize_value)(&val)
}
}
pub impl<D:Decoder,T:Decodable<D>> DeserializingUnflattener<D, T> {
fn new(deserialize_buffer: DeserializeBuffer<T>)
-> DeserializingUnflattener<D, T> {
2012-11-25 16:12:21 -06:00
DeserializingUnflattener {
2013-02-15 01:30:30 -06:00
deserialize_buffer: deserialize_buffer
2012-11-25 16:12:21 -06:00
}
}
}
pub impl<S:Encoder,T:Encodable<S>> SerializingFlattener<S, T> {
fn new(serialize_value: SerializeValue<T>)
-> SerializingFlattener<S, T> {
2012-11-25 16:12:21 -06:00
SerializingFlattener {
2013-02-15 01:30:30 -06:00
serialize_value: serialize_value
2012-11-25 16:12:21 -06:00
}
}
}
/*
Implementations of the serialization functions required by
SerializingFlattener
*/
pub fn deserialize_buffer<D: Decoder + FromReader,
T: Decodable<D>>(buf: &[u8]) -> T {
2012-11-25 16:12:21 -06:00
let buf = vec::from_slice(buf);
2013-02-15 01:30:30 -06:00
let buf_reader = @BufReader::new(buf);
2012-11-25 16:12:21 -06:00
let reader = buf_reader as @Reader;
let deser: D = FromReader::from_reader(reader);
Decodable::decode(&deser)
2012-11-25 16:12:21 -06:00
}
pub fn serialize_value<D: Encoder + FromWriter,
T: Encodable<D>>(val: &T) -> ~[u8] {
2013-03-07 22:44:38 -06:00
do io::with_bytes_writer |writer| {
let ser = FromWriter::from_writer(writer);
val.encode(&ser);
}
2012-11-25 16:12:21 -06:00
}
pub trait FromReader {
fn from_reader(r: @Reader) -> Self;
2012-11-25 16:12:21 -06:00
}
pub trait FromWriter {
fn from_writer(w: @Writer) -> Self;
2012-11-25 16:12:21 -06:00
}
impl<'self> FromReader for json::Decoder<'self> {
fn from_reader(r: @Reader) -> json::Decoder<'self> {
2012-11-25 16:12:21 -06:00
match json::from_reader(r) {
2013-02-15 01:30:30 -06:00
Ok(json) => {
json::Decoder(json)
2012-11-25 16:12:21 -06:00
}
Err(e) => fail!(fmt!("flatpipe: can't parse json: %?", e))
2012-11-25 16:12:21 -06:00
}
}
}
impl FromWriter for json::Encoder {
fn from_writer(w: @Writer) -> json::Encoder {
2013-02-15 01:30:30 -06:00
json::Encoder(w)
2012-11-25 16:12:21 -06:00
}
}
impl FromReader for ebml::reader::Decoder {
fn from_reader(r: @Reader) -> ebml::reader::Decoder {
2012-11-25 16:12:21 -06:00
let buf = @r.read_whole_stream();
let doc = ebml::reader::Doc(buf);
2013-02-15 01:30:30 -06:00
ebml::reader::Decoder(doc)
2012-11-25 16:12:21 -06:00
}
}
impl FromWriter for ebml::writer::Encoder {
fn from_writer(w: @Writer) -> ebml::writer::Encoder {
2013-02-15 01:30:30 -06:00
ebml::writer::Encoder(w)
2012-11-25 16:12:21 -06:00
}
}
}
pub mod bytepipes {
use flatpipes::{ByteChan, BytePort};
2012-11-25 16:12:21 -06:00
use core::io::{Writer, Reader, ReaderUtil};
2013-02-02 05:10:12 -06:00
use core::comm::{Port, Chan};
use core::prelude::*;
2012-11-25 16:12:21 -06:00
pub struct ReaderBytePort<R> {
2012-11-25 16:12:21 -06:00
reader: R
}
pub struct WriterByteChan<W> {
2012-11-25 16:12:21 -06:00
writer: W
}
impl<R:Reader> BytePort for ReaderBytePort<R> {
2012-11-25 16:12:21 -06:00
fn try_recv(&self, count: uint) -> Option<~[u8]> {
let mut left = count;
let mut bytes = ~[];
while !self.reader.eof() && left > 0 {
fail_unless!(left <= count);
fail_unless!(left > 0);
2012-11-25 16:12:21 -06:00
let new_bytes = self.reader.read_bytes(left);
bytes.push_all(new_bytes);
fail_unless!(new_bytes.len() <= left);
2012-11-25 16:12:21 -06:00
left -= new_bytes.len();
}
if left == 0 {
2013-02-15 01:30:30 -06:00
return Some(bytes);
2012-11-25 16:12:21 -06:00
} else {
warn!("flatpipe: dropped %? broken bytes", left);
return None;
}
}
}
impl<W:Writer> ByteChan for WriterByteChan<W> {
2012-11-25 16:12:21 -06:00
fn send(&self, val: ~[u8]) {
self.writer.write(val);
}
}
pub impl<R:Reader> ReaderBytePort<R> {
fn new(r: R) -> ReaderBytePort<R> {
2012-11-25 16:12:21 -06:00
ReaderBytePort {
2013-02-15 01:30:30 -06:00
reader: r
2012-11-25 16:12:21 -06:00
}
}
}
pub impl<W:Writer> WriterByteChan<W> {
fn new(w: W) -> WriterByteChan<W> {
2012-11-25 16:12:21 -06:00
WriterByteChan {
2013-02-15 01:30:30 -06:00
writer: w
2012-11-25 16:12:21 -06:00
}
}
}
pub struct PipeBytePort {
2013-02-02 05:10:12 -06:00
port: comm::Port<~[u8]>,
2012-11-25 16:12:21 -06:00
mut buf: ~[u8]
}
pub struct PipeByteChan {
2013-02-02 05:10:12 -06:00
chan: comm::Chan<~[u8]>
2012-11-25 16:12:21 -06:00
}
impl BytePort for PipeBytePort {
2012-11-25 16:12:21 -06:00
fn try_recv(&self, count: uint) -> Option<~[u8]> {
if vec::uniq_len(&const self.buf) >= count {
let mut bytes = ::core::util::replace(&mut self.buf, ~[]);
self.buf = bytes.slice(count, bytes.len()).to_owned();
2012-11-25 16:12:21 -06:00
bytes.truncate(count);
return Some(bytes);
} else if vec::uniq_len(&const self.buf) > 0 {
let mut bytes = ::core::util::replace(&mut self.buf, ~[]);
fail_unless!(count > bytes.len());
2012-11-25 16:12:21 -06:00
match self.try_recv(count - bytes.len()) {
2013-02-15 01:30:30 -06:00
Some(rest) => {
2012-11-25 16:12:21 -06:00
bytes.push_all(rest);
2013-02-15 01:30:30 -06:00
return Some(bytes);
2012-11-25 16:12:21 -06:00
}
None => return None
}
} else if vec::uniq_len(&const self.buf) == 0 {
2012-11-25 16:12:21 -06:00
match self.port.try_recv() {
2013-02-15 01:30:30 -06:00
Some(buf) => {
fail_unless!(!buf.is_empty());
2013-02-15 01:30:30 -06:00
self.buf = buf;
2012-11-25 16:12:21 -06:00
return self.try_recv(count);
}
None => return None
}
} else {
::core::util::unreachable()
2012-11-25 16:12:21 -06:00
}
}
}
impl ByteChan for PipeByteChan {
2012-11-25 16:12:21 -06:00
fn send(&self, val: ~[u8]) {
2013-02-15 01:30:30 -06:00
self.chan.send(val)
2012-11-25 16:12:21 -06:00
}
}
pub impl PipeBytePort {
fn new(p: Port<~[u8]>) -> PipeBytePort {
2012-11-25 16:12:21 -06:00
PipeBytePort {
2013-02-15 01:30:30 -06:00
port: p,
2012-11-25 16:12:21 -06:00
buf: ~[]
}
}
}
pub impl PipeByteChan {
fn new(c: Chan<~[u8]>) -> PipeByteChan {
2012-11-25 16:12:21 -06:00
PipeByteChan {
2013-02-15 01:30:30 -06:00
chan: c
2012-11-25 16:12:21 -06:00
}
}
}
}
#[cfg(test)]
mod test {
use core::prelude::*;
2012-11-25 16:12:21 -06:00
use flatpipes::{Flattener, Unflattener};
2012-11-25 16:12:21 -06:00
use flatpipes::bytepipes::*;
use flatpipes::pod;
use flatpipes::serial;
2013-01-10 22:09:26 -06:00
use io_util::BufReader;
2013-02-26 23:10:03 -06:00
use flatpipes::{BytePort, FlatChan, FlatPort};
use net::tcp::TcpSocketBuf;
2012-11-25 16:12:21 -06:00
use core::int;
2013-03-26 15:38:07 -05:00
use core::io::BytesWriter;
use core::prelude::*;
2012-12-28 14:46:08 -06:00
use core::result;
use core::task;
2012-11-25 16:12:21 -06:00
#[test]
fn test_serializing_memory_stream() {
let writer = BytesWriter();
2013-02-15 01:30:30 -06:00
let chan = serial::writer_chan(writer);
2012-11-25 16:12:21 -06:00
chan.send(10);
2013-03-07 22:44:38 -06:00
let bytes = copy chan.byte_chan.writer.bytes;
2012-11-25 16:12:21 -06:00
2013-02-15 01:30:30 -06:00
let reader = BufReader::new(bytes);
let port = serial::reader_port(reader);
2012-11-25 16:12:21 -06:00
let res: int = port.recv();
fail_unless!(res == 10i);
2012-11-25 16:12:21 -06:00
}
#[test]
fn test_serializing_pipes() {
let (port, chan) = serial::pipe_stream();
2013-02-15 01:30:30 -06:00
do task::spawn || {
2012-11-25 16:12:21 -06:00
for int::range(0, 10) |i| {
chan.send(i)
}
}
for int::range(0, 10) |i| {
fail_unless!(i == port.recv())
2012-11-25 16:12:21 -06:00
}
}
#[test]
fn test_serializing_boxes() {
let (port, chan) = serial::pipe_stream();
2013-02-15 01:30:30 -06:00
do task::spawn || {
2012-11-25 16:12:21 -06:00
for int::range(0, 10) |i| {
chan.send(@i)
}
}
for int::range(0, 10) |i| {
fail_unless!(@i == port.recv())
2012-11-25 16:12:21 -06:00
}
}
#[test]
fn test_pod_memory_stream() {
let writer = BytesWriter();
2013-02-15 01:30:30 -06:00
let chan = pod::writer_chan(writer);
2012-11-25 16:12:21 -06:00
chan.send(10);
2013-03-07 22:44:38 -06:00
let bytes = copy chan.byte_chan.writer.bytes;
2012-11-25 16:12:21 -06:00
2013-02-15 01:30:30 -06:00
let reader = BufReader::new(bytes);
let port = pod::reader_port(reader);
2012-11-25 16:12:21 -06:00
let res: int = port.recv();
fail_unless!(res == 10);
2012-11-25 16:12:21 -06:00
}
#[test]
fn test_pod_pipes() {
let (port, chan) = pod::pipe_stream();
2013-02-15 01:30:30 -06:00
do task::spawn || {
2012-11-25 16:12:21 -06:00
for int::range(0, 10) |i| {
chan.send(i)
}
}
for int::range(0, 10) |i| {
fail_unless!(i == port.recv())
2012-11-25 16:12:21 -06:00
}
}
2013-01-10 22:09:26 -06:00
// FIXME #2064: Networking doesn't work on x86
2012-11-25 16:12:21 -06:00
#[test]
#[cfg(target_arch = "x86_64")]
fn test_pod_tcp_stream() {
fn reader_port(buf: TcpSocketBuf
) -> pod::ReaderPort<int, TcpSocketBuf> {
2013-02-15 01:30:30 -06:00
pod::reader_port(buf)
2012-11-25 16:12:21 -06:00
}
fn writer_chan(buf: TcpSocketBuf
) -> pod::WriterChan<int, TcpSocketBuf> {
2013-02-15 01:30:30 -06:00
pod::writer_chan(buf)
2012-11-25 16:12:21 -06:00
}
test_some_tcp_stream(reader_port, writer_chan, 9666);
}
#[test]
#[cfg(target_arch = "x86_64")]
fn test_serializing_tcp_stream() {
fn reader_port(buf: TcpSocketBuf
) -> serial::ReaderPort<int, TcpSocketBuf> {
2013-02-15 01:30:30 -06:00
serial::reader_port(buf)
2012-11-25 16:12:21 -06:00
}
fn writer_chan(buf: TcpSocketBuf
) -> serial::WriterChan<int, TcpSocketBuf> {
2013-02-15 01:30:30 -06:00
serial::writer_chan(buf)
2012-11-25 16:12:21 -06:00
}
test_some_tcp_stream(reader_port, writer_chan, 9667);
}
type ReaderPortFactory<U> =
2012-11-25 16:12:21 -06:00
~fn(TcpSocketBuf) -> FlatPort<int, U, ReaderBytePort<TcpSocketBuf>>;
type WriterChanFactory<F> =
2012-11-25 16:12:21 -06:00
~fn(TcpSocketBuf) -> FlatChan<int, F, WriterByteChan<TcpSocketBuf>>;
fn test_some_tcp_stream<U:Unflattener<int>,F:Flattener<int>>(
2012-11-25 16:12:21 -06:00
reader_port: ReaderPortFactory<U>,
writer_chan: WriterChanFactory<F>,
port: uint) {
use core::cell::Cell;
2012-11-25 16:12:21 -06:00
use net::ip;
use net::tcp;
2012-12-28 14:46:08 -06:00
use uv;
2012-11-25 16:12:21 -06:00
// Indicate to the client task that the server is listening
2013-02-02 05:10:12 -06:00
let (begin_connect_port, begin_connect_chan) = comm::stream();
2012-11-25 16:12:21 -06:00
// The connection is sent from the server task to the receiver task
// to handle the connection
2013-02-02 05:10:12 -06:00
let (accept_port, accept_chan) = comm::stream();
2012-11-25 16:12:21 -06:00
// The main task will wait until the test is over to proceed
2013-02-02 05:10:12 -06:00
let (finish_port, finish_chan) = comm::stream();
2012-11-25 16:12:21 -06:00
let addr0 = ip::v4::parse_addr("127.0.0.1");
2012-11-25 16:12:21 -06:00
2013-02-15 01:30:30 -06:00
let begin_connect_chan = Cell(begin_connect_chan);
let accept_chan = Cell(accept_chan);
2012-11-25 16:12:21 -06:00
// The server task
let addr = copy addr0;
2013-02-15 01:30:30 -06:00
do task::spawn || {
let iotask = &uv::global_loop::get();
2012-11-25 16:12:21 -06:00
let begin_connect_chan = begin_connect_chan.take();
let accept_chan = accept_chan.take();
let listen_res = do tcp::listen(
2013-02-15 01:30:30 -06:00
copy addr, port, 128, iotask, |_kill_ch| {
2012-11-25 16:12:21 -06:00
// Tell the sender to initiate the connection
debug!("listening");
begin_connect_chan.send(())
2013-02-15 01:30:30 -06:00
}) |new_conn, kill_ch| {
2012-11-25 16:12:21 -06:00
// Incoming connection. Send it to the receiver task to accept
2013-02-02 05:10:12 -06:00
let (res_port, res_chan) = comm::stream();
2013-02-15 01:30:30 -06:00
accept_chan.send((new_conn, res_chan));
2012-11-25 16:12:21 -06:00
// Wait until the connection is accepted
res_port.recv();
// Stop listening
kill_ch.send(None)
};
fail_unless!(listen_res.is_ok());
2012-11-25 16:12:21 -06:00
}
// Client task
let addr = copy addr0;
2013-02-15 01:30:30 -06:00
do task::spawn || {
2012-11-25 16:12:21 -06:00
// Wait for the server to start listening
begin_connect_port.recv();
debug!("connecting");
let iotask = &uv::global_loop::get();
2012-11-25 16:12:21 -06:00
let connect_result = tcp::connect(copy addr, port, iotask);
fail_unless!(connect_result.is_ok());
2013-02-15 01:30:30 -06:00
let sock = result::unwrap(connect_result);
let socket_buf: tcp::TcpSocketBuf = tcp::socket_buf(sock);
2012-11-25 16:12:21 -06:00
// TcpSocketBuf is a Writer!
2013-02-15 01:30:30 -06:00
let chan = writer_chan(socket_buf);
2012-11-25 16:12:21 -06:00
for int::range(0, 10) |i| {
debug!("sending %?", i);
chan.send(i)
}
}
// Reciever task
2013-02-15 01:30:30 -06:00
do task::spawn || {
2012-11-25 16:12:21 -06:00
// Wait for a connection
let (conn, res_chan) = accept_port.recv();
debug!("accepting connection");
let accept_result = tcp::accept(conn);
debug!("accepted");
fail_unless!(accept_result.is_ok());
2013-02-15 01:30:30 -06:00
let sock = result::unwrap(accept_result);
2012-11-25 16:12:21 -06:00
res_chan.send(());
2013-02-15 01:30:30 -06:00
let socket_buf: tcp::TcpSocketBuf = tcp::socket_buf(sock);
2012-11-25 16:12:21 -06:00
// TcpSocketBuf is a Reader!
2013-02-15 01:30:30 -06:00
let port = reader_port(socket_buf);
2012-11-25 16:12:21 -06:00
for int::range(0, 10) |i| {
let j = port.recv();
debug!("receieved %?", j);
fail_unless!(i == j);
2012-11-25 16:12:21 -06:00
}
// The test is over!
finish_chan.send(());
}
finish_port.recv();
}
// Tests that the different backends behave the same when the
// binary streaming protocol is broken
mod broken_protocol {
use core::prelude::*;
use flatpipes::{BytePort, FlatPort};
use flatpipes::flatteners::PodUnflattener;
2012-12-28 14:46:08 -06:00
use flatpipes::pod;
2013-01-10 22:09:26 -06:00
use io_util::BufReader;
2012-12-28 14:46:08 -06:00
use core::io;
use core::sys;
use core::task;
type PortLoader<P> =
2012-11-25 16:12:21 -06:00
~fn(~[u8]) -> FlatPort<int, PodUnflattener<int>, P>;
fn reader_port_loader(bytes: ~[u8]
) -> pod::ReaderPort<int, BufReader> {
2013-02-15 01:30:30 -06:00
let reader = BufReader::new(bytes);
pod::reader_port(reader)
2012-11-25 16:12:21 -06:00
}
fn pipe_port_loader(bytes: ~[u8]
) -> pod::PipePort<int> {
2013-02-02 05:10:12 -06:00
let (port, chan) = comm::stream();
if !bytes.is_empty() {
2013-02-15 01:30:30 -06:00
chan.send(bytes);
2012-11-25 16:12:21 -06:00
}
2013-02-15 01:30:30 -06:00
pod::pipe_port(port)
2012-11-25 16:12:21 -06:00
}
fn test_try_recv_none1<P:BytePort>(loader: PortLoader<P>) {
2012-11-25 16:12:21 -06:00
let bytes = ~[];
2013-02-15 01:30:30 -06:00
let port = loader(bytes);
2012-11-25 16:12:21 -06:00
let res: Option<int> = port.try_recv();
fail_unless!(res.is_none());
2012-11-25 16:12:21 -06:00
}
#[test]
fn test_try_recv_none1_reader() {
test_try_recv_none1(reader_port_loader);
}
#[test]
fn test_try_recv_none1_pipe() {
test_try_recv_none1(pipe_port_loader);
}
fn test_try_recv_none2<P:BytePort>(loader: PortLoader<P>) {
2012-11-25 16:12:21 -06:00
// The control word in the protocol is interrupted
let bytes = ~[0];
2013-02-15 01:30:30 -06:00
let port = loader(bytes);
2012-11-25 16:12:21 -06:00
let res: Option<int> = port.try_recv();
fail_unless!(res.is_none());
2012-11-25 16:12:21 -06:00
}
#[test]
fn test_try_recv_none2_reader() {
test_try_recv_none2(reader_port_loader);
}
#[test]
fn test_try_recv_none2_pipe() {
test_try_recv_none2(pipe_port_loader);
}
fn test_try_recv_none3<P:BytePort>(loader: PortLoader<P>) {
static CONTINUE: [u8, ..4] = [0xAA, 0xBB, 0xCC, 0xDD];
2012-11-25 16:12:21 -06:00
// The control word is followed by garbage
let bytes = CONTINUE.to_vec() + ~[0];
2013-02-15 01:30:30 -06:00
let port = loader(bytes);
2012-11-25 16:12:21 -06:00
let res: Option<int> = port.try_recv();
fail_unless!(res.is_none());
2012-11-25 16:12:21 -06:00
}
#[test]
fn test_try_recv_none3_reader() {
test_try_recv_none3(reader_port_loader);
}
#[test]
fn test_try_recv_none3_pipe() {
test_try_recv_none3(pipe_port_loader);
}
fn test_try_recv_none4<P:BytePort>(+loader: PortLoader<P>) {
fail_unless!(do task::try || {
static CONTINUE: [u8, ..4] = [0xAA, 0xBB, 0xCC, 0xDD];
2012-11-25 16:12:21 -06:00
// The control word is followed by a valid length,
// then undeserializable garbage
let len_bytes = do io::u64_to_be_bytes(
1, sys::size_of::<u64>()) |len_bytes| {
len_bytes.to_vec()
};
let bytes = CONTINUE.to_vec() + len_bytes + ~[0, 0, 0, 0];
2013-02-15 01:30:30 -06:00
let port = loader(bytes);
2012-11-25 16:12:21 -06:00
let _res: Option<int> = port.try_recv();
}.is_err());
2012-11-25 16:12:21 -06:00
}
#[test]
#[ignore(cfg(windows))]
fn test_try_recv_none4_reader() {
test_try_recv_none4(reader_port_loader);
}
#[test]
#[ignore(cfg(windows))]
fn test_try_recv_none4_pipe() {
test_try_recv_none4(pipe_port_loader);
}
}
}