1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
// Copyright 2015 Nathan Sizemore <nathanrsizemore@gmail.com>
//
// This Source Code Form is subject to the terms of the
// Mozilla Public License, v. 2.0. If a copy of the MPL was not
// distributed with this file, You can obtain one at
// http://mozilla.org/MPL/2.0/.


//! hydrogen is a non-blocking Edge Triggered TCP socket lib built atop [epoll][epoll-man-page]
//! with performance, concurrency, and scalability as its main priorities. It takes care of the
//! tedious connection and I/O marshalling across threads, and leaves the specifics of I/O reading
//! and writing up the consumer, through trait implementations.
//!
//! # Streams
//!
//! hydrogen manages the state of connections through [`hydrogen::Stream`][stream-trait]
//! [Trait Objects][trait-objects].
//!
//! # Events
//!
//! hydrogen reports all events to the [`hydrogen::Handler`][handler] passed during creation.
//! Interaction to `hydrogen::Stream` trait objects is made through a simple wrapper,
//! `HydrogenSocket`, to ensure thread safety.
//!
//! # Example Usage
//!
//! The following is a simple snippet using the [simple-stream][simple-stream-repo] crate to
//! provide the non-blocking I/O calls.
//!
//!
//! ```ignore
//! extern crate hydrogen;
//! extern crate simple_stream as ss;
//!
//! use hydrogen;
//! use hydrogen::{Stream as HydrogenStream, HydrogenSocket};
//! use ss::frame::Frame;
//! use ss::frame::simple::{SimpleFrame, SimpleFrameBuilder};
//! use ss::{Socket, Plain, NonBlocking, SocketOptions};
//!
//!
//! #[derive(Clone)]
//! pub struct Stream {
//!     inner: Plain<Socket, SimpleFrameBuilder>
//! }
//!
//! impl HydrogenStream for Stream {
//!     fn recv(&mut self) -> Result<Vec<Vec<u8>>, Error> {
//!         match self.inner.nb_recv() {
//!             Ok(frame_vec) => {
//!                 let mut ret_buf = Vec::<Vec<u8>>::with_capacity(frame_vec.len());
//!                 for frame in frame_vec.iter() {
//!                     ret_buf.push(frame.payload());
//!                 }
//!                 Ok(ret_buf)
//!             }
//!             Err(e) => Err(e)
//!         }
//!     }
//!
//!     fn send(&mut self, buf: &[u8]) -> Result<(), Error> {
//!         let frame = SimpleFrame::new(buf);
//!         self.inner.nb_send(&frame)
//!     }
//!
//!     fn shutdown(&mut self) -> Result<(), Error> {
//!         self.inner.shutdown()
//!     }
//! }
//! impl AsRawFd for Stream {
//!     fn as_raw_fd(&self) -> RawFd { self.inner.as_raw_fd() }
//! }
//!
//!
//! struct Server;
//! impl hydrogen::Handler for Server {
//!     fn on_server_created(&mut self, fd: RawFd) {
//!
//!     }
//!
//!     fn on_new_connection(&mut self, fd: RawFd) -> Arc<UnsafeCell<HydrogenStream>> {
//!
//!     }
//!
//!     fn on_data_received(&mut self, socket: HydrogenSocket, buffer: Vec<u8>) {
//!
//!     }
//!
//!     fn on_connection_removed(&mut self, fd: RawFd, err: Error) {
//!
//!     }
//! }
//!
//!
//! fn main() {
//!     hydrogen::begin(Server, hydrogen::Config {
//!         addr: "0.0.0.0".to_string(),
//!         port: 1337,
//!         max_threads: 8,
//!         pre_allocated: 100000
//!     });
//! }
//!
//! ```
//!
//!
//!
//! [epoll-man-page]: http://man7.org/linux/man-pages/man7/epoll.7.html
//! [stream-trait]: https://nathansizemore.github.io/hydrogen/hydrogen/trait.Stream.html
//! [trait-objects]: https://doc.rust-lang.org/book/trait-objects.html
//! [handler]: https://nathansizemore.github.io/hydrogen/hydrogen/trait.Handler.html
//! [simple-stream-repo]: https://github.com/nathansizemore/simple-stream


#[macro_use]
extern crate log;
extern crate libc;
extern crate errno;
extern crate threadpool;
extern crate simple_slab;


use std::io::Error;
use std::sync::Arc;
use std::cell::UnsafeCell;
use std::os::unix::io::{RawFd, AsRawFd};


pub use config::Config;
pub use types::HydrogenSocket;

mod types;
mod server;
mod config;


/// Trait object responsible for handling reported I/O events.
pub trait Stream : AsRawFd + Send + Sync {
    /// Called when epoll reports data is available for read.
    ///
    /// This method should read until `ErrorKind::WouldBlock` is received. At that time, all
    /// complete messages should be returned, otherwise return the std::io::Error.
    fn recv(&mut self) -> Result<Vec<Vec<u8>>, Error>;
    /// Called as the internal writer for the HydrogenSocket wrapper.
    ///
    /// This method should write until all bytes have been written or any `std::io::Error` is
    /// returned.
    fn send(&mut self, buf: &[u8]) -> Result<(), Error>;
    /// This method is called when any error, other than `ErrorKind::WouldBlock`, is returned from
    /// a `recv` or `send` call.
    fn shutdown(&mut self) -> Result<(), Error>;
}

/// Events reported to lib consumer.
pub trait Handler {
    /// This method is called once the listening RawFd has been created.
    ///
    /// It should be used to set/remove any flags on the underlying RawFd before `listen` is
    /// called on the fd.
    fn on_server_created(&mut self, fd: RawFd);
    /// This method is called whenever `accept` returns a new TCP connection.
    ///
    /// The returned trait object is added to the connection pool and the epoll interest list.
    fn on_new_connection(&mut self, fd: RawFd) -> Arc<UnsafeCell<Stream>>;
    /// This method is called whenever the `recv` call returns an Ok(_) result.
    fn on_data_received(&mut self, socket: HydrogenSocket, buf: Vec<u8>);
    /// This method is called after a stream has been removed from the connection poll and epoll
    /// interest list, with the `std::io::Error` as the reason removed.
    ///
    /// At the time of this call, the underlying fd has been shutdown and closed. No system level
    /// shutdown is needed, only application level cleanup.
    fn on_connection_removed(&mut self, fd: RawFd, err: Error);
}

/// Starts the server with the passed configuration and handler.
pub fn begin<T>(handler: Box<T>, cfg: Config)
    where T: Handler + Send + Sync + 'static
{
    server::begin(handler, cfg);
}