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); }