flute/lib.rs
1//! [](https://github.com/ypo/flute/actions/workflows/rust.yml)
2//! [](https://github.com/ypo/flute/actions/workflows/python.yml)
3//! [](https://docs.rs/crate/flute/)
4//! [](https://crates.io/crates/flute/)
5//! [](https://deps.rs/repo/github/ypo/flute)
6//! [](https://codecov.io/gh/ypo/flute)
7//!
8//! # FLUTE - File Delivery over Unidirectional Transport
9//!
10//!
11//! Massively scalable multicast distribution solution
12//!
13//! The library implements a unidirectional file delivery, without the need of a return channel.
14//!
15//!
16//! # RFC
17//!
18//! This library implements the following RFCs
19//!
20//!| RFC | Title | Link |
21//!| -------- | ---------- | -----------|
22//!| RFC 6726 | FLUTE - File Delivery over Unidirectional Transport | <https://www.rfc-editor.org/rfc/rfc6726.html> |
23//!| RFC 5775 | Asynchronous Layered Coding (ALC) Protocol Instantiation | <https://www.rfc-editor.org/rfc/rfc5775.html> |
24//!| RFC 5661 | Layered Coding Transport (LCT) Building Block | <https://www.rfc-editor.org/rfc/rfc5651> |
25//!| RFC 5052 | Forward Error Correction (FEC) Building Block | <https://www.rfc-editor.org/rfc/rfc5052> |
26//!| RFC 5510 | Reed-Solomon Forward Error Correction (FEC) Schemes | <https://www.rfc-editor.org/rfc/rfc5510.html> |
27//!| 3GPP TS 26.346 | Extended FLUTE FDT Schema (7.2.10) | <https://www.etsi.org/deliver/etsi_ts/126300_126399/126346/17.03.00_60/ts_126346v170300p.pdf> |
28//!
29//! # Thread Safety
30//!
31//! ## FLUTE Sender
32//!
33//! The FLUTE Sender is designed to be safely shared between multiple threads.
34//!
35//! ## FLUTE Receiver and Tokio Integration
36//!
37//! Unlike the sender, the FLUTE Receiver **is not thread-safe** and cannot be shared between multiple threads.
38//! To integrate it with Tokio, you must use `tokio::task::LocalSet`, which allows spawning tasks that require a single-threaded runtime.
39//!
40//! The following example demonstrates how to use the FLUTE Receiver with Tokio:
41//!
42//!```ignore
43//! use flute::receiver::{writer, MultiReceiver};
44//! use std::rc::Rc;
45//!
46//!#[tokio::main]
47//!async fn main() {
48//! let local = task::LocalSet::new();
49//! // Run the local task set.
50//! local.run_until(async move {
51//! let nonsend_data = nonsend_data.clone();
52//! task::spawn_local(async move {
53//! let writer = Rc::new(writer::ObjectWriterFSBuilder::new(&std::path::Path::new("./flute_dir"), true).unwrap_or_else(|_| std::process::exit(0)));
54//! let mut receiver = MultiReceiver::new(writer, None, false);
55//! // ... run the receiver
56//! }).await.unwrap();
57//! }).await;
58//!}
59//!```
60//! # UDP/IP Multicast files sender
61//!
62//! Transfer files over a UDP/IP network
63//!
64//!```rust
65//! use flute::sender::Sender;
66//! use flute::sender::ObjectDesc;
67//! use flute::core::lct::Cenc;
68//! use flute::core::UDPEndpoint;
69//! use std::net::UdpSocket;
70//! use std::time::SystemTime;
71//!
72//! // Create UDP Socket
73//! let udp_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
74//! udp_socket.connect("224.0.0.1:3400").expect("Connection failed");
75//!
76//! // Create FLUTE Sender
77//! let tsi = 1;
78//! let oti = Default::default();
79//! let config = Default::default();
80//! let endpoint = UDPEndpoint::new(None, "224.0.0.1".to_string(), 3400);
81//! let mut sender = Sender::new(endpoint, tsi, &oti, &config);
82//!
83//! // Add object(s) (files) to the FLUTE sender (priority queue 0)
84//! let obj = ObjectDesc::create_from_buffer(b"hello world".to_vec(), "text/plain",
85//! &url::Url::parse("file:///hello.txt").unwrap(), 1, None, None, None, None, Cenc::Null, true, None, true).unwrap();
86//! sender.add_object(0, obj);
87//!
88//! // Always call publish after adding objects
89//! sender.publish(SystemTime::now());
90//!
91//! // Send FLUTE packets over UDP/IP
92//! while let Some(pkt) = sender.read(SystemTime::now()) {
93//! udp_socket.send(&pkt).unwrap();
94//! std::thread::sleep(std::time::Duration::from_millis(1));
95//! }
96//!
97//!```
98//! # UDP/IP Multicast files receiver
99//!
100//! Receive files from a UDP/IP network
101//!
102//!```
103//! use flute::receiver::{writer, MultiReceiver};
104//! use flute::core::UDPEndpoint;
105//! use std::net::UdpSocket;
106//! use std::time::SystemTime;
107//! use std::rc::Rc;
108//!
109//! // Create UDP/IP socket to receive FLUTE pkt
110//! let endpoint = UDPEndpoint::new(None, "224.0.0.1".to_string(), 3400);
111//! let udp_socket = UdpSocket::bind(format!("{}:{}", endpoint.destination_group_address, endpoint.port)).expect("Fail to bind");
112//!
113//! // Create a writer able to write received files to the filesystem
114//! let enable_md5_check = true;
115//! let writer = Rc::new(writer::ObjectWriterFSBuilder::new(&std::path::Path::new("./flute_dir"), enable_md5_check)
116//! .unwrap_or_else(|_| std::process::exit(0)));
117//!
118//! // Create a multi-receiver capable of de-multiplexing several FLUTE sessions
119//! let mut receiver = MultiReceiver::new(writer, None, false);
120//!
121//! // Receive pkt from UDP/IP socket and push it to the FLUTE receiver
122//! let mut buf = [0; 2048];
123//! loop {
124//! let (n, _src) = udp_socket.recv_from(&mut buf).expect("Failed to receive data");
125//! let now = SystemTime::now();
126//! receiver.push(&endpoint, &buf[..n], now).unwrap();
127//! receiver.cleanup(now);
128//! }
129//!```
130//! # Application-Level Forward Erasure Correction (AL-FEC)
131//!
132//! The following error recovery algorithms are supported
133//!
134//! - [X] No-code
135//! - [X] Reed-Solomon GF 2^8
136//! - [X] Reed-Solomon GF 2^8 Under Specified
137//! - [ ] Reed-Solomon GF 2^16
138//! - [ ] Reed-Solomon GF 2^m
139//! - [X] RaptorQ
140//! - [X] Raptor
141//!
142//! The `Oti` module provides an implementation of the Object Transmission Information (OTI)
143//! used to configure Forward Error Correction (FEC) encoding in the FLUTE protocol.
144//!
145//!```rust
146//! use flute::sender::Sender;
147//! use flute::core::Oti;
148//! use flute::core::UDPEndpoint;
149//!
150//! // Reed Solomon 2^8 with encoding blocks composed of
151//! // 60 source symbols and 4 repair symbols of 1424 bytes per symbol
152//! let endpoint = UDPEndpoint::new(None, "224.0.0.1".to_string(), 3400);
153//! let oti = Oti::new_reed_solomon_rs28(1424, 60, 4).unwrap();
154//! let mut sender = Sender::new(endpoint, 1, &oti, &Default::default());
155//!```
156//!
157//! # Content Encoding (CENC)
158//!
159//! The following schemes are supported during the transmission/reception
160//!
161//! - [x] Null (no compression)
162//! - [x] Deflate
163//! - [x] Zlib
164//! - [x] Gzip
165//!
166//! # Files multiplex / Blocks interleave
167//!
168//! The FLUTE Sender is able to transfer multiple files in parallel by interleaving packets from each file. For example:
169//!
170//! **Pkt file1** -> Pkt file2 -> Pkt file3 -> **Pkt file1** -> Pkt file2 -> Pkt file3 ...
171//!
172//! The Sender can interleave blocks within a single file.
173//! The following example shows Encoding Symbols (ES) from different blocks (B) are interleaved. For example:
174//!
175//! **(B 1,ES 1)**->(B 2,ES 1)->(B 3,ES 1)->**(B 1,ES 2)**->(B 2,ES 2)...
176//!
177//! To configure the multiplexing, use the `Config` struct as follows:
178//!
179//!```rust
180//! use flute::sender::Sender;
181//! use flute::sender::Config;
182//! use flute::sender::PriorityQueue;
183//! use flute::core::UDPEndpoint;
184//!
185//! let mut config = Config {
186//! // Interleave a maximum of 3 blocks within each file
187//! interleave_blocks: 3,
188//! ..Default::default()
189//! };
190//!
191//! // Interleave a maximum of 3 files in priority queue '0'
192//! config.set_priority_queue(PriorityQueue::HIGHEST, PriorityQueue::new(3));
193//!
194//! let endpoint = UDPEndpoint::new(None, "224.0.0.1".to_string(), 3400);
195//! let mut sender = Sender::new(endpoint, 1, &Default::default(), &config);
196//!```
197//!
198//! # Priority Queues
199//!
200//! FLUTE sender can be configured with multiple queues, each having a different priority level.
201//! Files in higher priority queues are always transferred before files in lower priority queues.
202//! Transfer of files in lower priority queues is paused while there are files to be transferred in higher priority queues.
203//!
204//!```rust
205//!
206//! use flute::sender::Sender;
207//! use flute::sender::Config;
208//! use flute::sender::PriorityQueue;
209//! use flute::core::UDPEndpoint;
210//! use flute::sender::ObjectDesc;
211//! use flute::core::lct::Cenc;
212//!
213//! // Create a default configuration
214//! let mut config: flute::sender::Config = Default::default();
215//!
216//! // Configure the HIGHEST priority queue with a capacity of 3 simultaneous file transfer
217//! config.set_priority_queue(PriorityQueue::HIGHEST, PriorityQueue::new(3));
218//!
219//! // Configure the LOW priority queue with a capacity of 1 file transfer at a time
220//! config.set_priority_queue(PriorityQueue::LOW, PriorityQueue::new(1));
221//!
222//! let endpoint = UDPEndpoint::new(None, "224.0.0.1".to_string(), 3400);
223//! let mut sender = Sender::new(endpoint, 1, &Default::default(), &config);
224//!
225//! // Create an ObjectDesc for a low priority file
226//! let low_priority_obj = ObjectDesc::create_from_buffer(b"low priority".to_vec(), "text/plain",
227//! &url::Url::parse("file:///low_priority.txt").unwrap(), 1, None, None, None, None, Cenc::Null, true, None, true).unwrap();
228//!
229//! // Create an ObjectDesc for a high priority file
230//! let high_priority_obj = ObjectDesc::create_from_buffer(b"high priority".to_vec(), "text/plain",
231//! &url::Url::parse("file:///high_priority.txt").unwrap(), 1, None, None, None, None, Cenc::Null, true, None, true).unwrap();
232//!
233//! // Put Object to the low priority queue
234//! sender.add_object(PriorityQueue::LOW, low_priority_obj);
235//!
236//! // Put Object to the high priority queue
237//! sender.add_object(PriorityQueue::HIGHEST, high_priority_obj);
238//!```
239//!
240//! # Bitrate Control
241//!The FLUTE library does not provide a built-in bitrate control mechanism.
242//! Users are responsible for controlling the bitrate by sending packets at a specific rate.
243//! However, the library offers a way to control the target transfer duration or the target transfer end time for each file individually.
244//!
245//!To ensure proper functionality, the target transfer mechanism requires that the overall bitrate is sufficiently high.
246//!
247//! ## Target Transfer Duration
248//!
249//! The sender will attempt to transfer the file within the specified duration.
250//!
251//!```rust
252//!
253//! use flute::sender::Sender;
254//! use flute::sender::ObjectDesc;
255//! use flute::sender::TargetAcquisition;
256//! use flute::core::lct::Cenc;
257//! use flute::core::UDPEndpoint;
258//! use std::net::UdpSocket;
259//! use std::time::SystemTime;
260//!
261//! // Create UDP Socket
262//! let udp_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
263//! udp_socket.connect("224.0.0.1:3400").expect("Connection failed");
264//!
265//! // Create FLUTE Sender
266//! let tsi = 1;
267//! let oti = Default::default();
268//! let config = Default::default();
269//! let endpoint = UDPEndpoint::new(None, "224.0.0.1".to_string(), 3400);
270//! let mut sender = Sender::new(endpoint, tsi, &oti, &config);
271//!
272//! // Create an Object
273//! let mut obj = ObjectDesc::create_from_buffer(b"hello world".to_vec(), "text/plain",
274//! &url::Url::parse("file:///hello.txt").unwrap(), 1, None, None, None, None, Cenc::Null, true, None, true).unwrap();
275//!
276//! // Set the Target Transfer Duration of this object to 2 seconds
277//! obj.target_acquisition = Some(TargetAcquisition::WithinDuration(std::time::Duration::from_secs(2)));
278//!
279//! // Add object(s) (files) to the FLUTE sender (priority queue 0)
280//! sender.add_object(0, obj);
281//!
282//! // Always call publish after adding objects
283//! sender.publish(SystemTime::now());
284//!
285//! // Send FLUTE packets over UDP/IP
286//! while sender.nb_objects() > 0 {
287//! if let Some(pkt) = sender.read(SystemTime::now()) {
288//! udp_socket.send(&pkt).unwrap();
289//! } else {
290//! std::thread::sleep(std::time::Duration::from_millis(1));
291//! }
292//! }
293//!```
294//!
295//! ## Target Time to End Transfer
296//!
297//! The sender will try to finish the file at the specified time.
298//!
299//!```rust
300//!
301//! use flute::sender::Sender;
302//! use flute::sender::ObjectDesc;
303//! use flute::sender::TargetAcquisition;
304//! use flute::core::lct::Cenc;
305//! use flute::core::UDPEndpoint;
306//! use std::net::UdpSocket;
307//! use std::time::SystemTime;
308//!
309//! // Create UDP Socket
310//! let udp_socket = UdpSocket::bind("0.0.0.0:0").unwrap();
311//! udp_socket.connect("224.0.0.1:3400").expect("Connection failed");
312//!
313//! // Create FLUTE Sender
314//! let tsi = 1;
315//! let oti = Default::default();
316//! let config = Default::default();
317//! let endpoint = UDPEndpoint::new(None, "224.0.0.1".to_string(), 3400);
318//! let mut sender = Sender::new(endpoint, tsi, &oti, &config);
319//!
320//! // Create an Object
321//! let mut obj = ObjectDesc::create_from_buffer(b"hello world".to_vec(), "text/plain",
322//! &url::Url::parse("file:///hello.txt").unwrap(), 1, None, None, None, None, Cenc::Null, true, None, true).unwrap();
323//!
324//! // Set the Target Transfer End Time of this object to 10 seconds from now
325//! let target_end_time = SystemTime::now() + std::time::Duration::from_secs(10);
326//! obj.target_acquisition = Some(TargetAcquisition::WithinTime(target_end_time));
327//!
328//! // Add object(s) (files) to the FLUTE sender (priority queue 0)
329//! sender.add_object(0, obj);
330//!
331//! // Always call publish after adding objects
332//! sender.publish(SystemTime::now());
333//!
334//! // Send FLUTE packets over UDP/IP
335//! while sender.nb_objects() > 0 {
336//! if let Some(pkt) = sender.read(SystemTime::now()) {
337//! udp_socket.send(&pkt).unwrap();
338//! } else {
339//! std::thread::sleep(std::time::Duration::from_millis(1));
340//! }
341//! }
342//!```
343
344#![deny(missing_docs)]
345#![deny(missing_debug_implementations)]
346#![cfg_attr(test, deny(warnings))]
347
348mod common;
349mod fec;
350mod tools;
351
352pub mod receiver;
353pub mod sender;
354pub use crate::tools::error;
355
356/// Core module with low-level function
357pub mod core {
358
359 /// ALC packets
360 pub mod alc {
361 pub use crate::common::alc::get_sender_current_time;
362 pub use crate::common::alc::parse_alc_pkt;
363 pub use crate::common::alc::parse_payload_id;
364 pub use crate::common::alc::AlcPkt;
365 pub use crate::common::alc::PayloadID;
366 }
367
368 /// LCT packets
369 pub mod lct {
370 pub use crate::common::lct::get_ext;
371 pub use crate::common::lct::inc_hdr_len;
372 pub use crate::common::lct::push_lct_header;
373 pub use crate::common::lct::Cenc;
374 pub use crate::common::lct::LCTHeader;
375 }
376
377 pub use crate::common::oti::FECEncodingID;
378 pub use crate::common::oti::Oti;
379 pub use crate::common::udpendpoint::UDPEndpoint;
380}
381
382#[cfg(feature = "python")]
383mod py;
384
385#[cfg(test)]
386mod tests {
387 pub fn init() {
388 std::env::set_var("RUST_LOG", "debug");
389 env_logger::builder().is_test(true).try_init().ok();
390 }
391}