Struct kekbit::core::ShmWriter

pub struct ShmWriter { /* fields omitted */ }

An implementation of the Writer which access a persistent channel through memory mapping. A ShmWriter must be created using the shm_writer function. Any ShmWriter exclusively holds the channel is bound to, and it is not thread safe. If multiple threads must write into a channel they should be externally synchronized.

Examples

use kekbit_core::tick::TickUnit::Nanos;
use kekbit_core::shm::*;
use kekbit_core::header::Header;
use kekbit_core::api::Writer;

const FOREVER: u64 = 99_999_999_999;
let writer_id = 1850;
let channel_id = 42;
let capacity = 3000;
let max_msg_len = 100;
let header = Header::new(writer_id, channel_id, capacity, max_msg_len, FOREVER, Nanos);
let test_tmp_dir = tempdir::TempDir::new("kektest").unwrap();
let mut writer = shm_writer(&test_tmp_dir.path(), &header).unwrap();
writer.heartbeat().unwrap();

Methods

impl ShmWriter

pub fn available(&self) -> u32

Returns the amount of space still available into this channel.

pub fn write_offset(&self) -> u32

Returns the amount of data written into this channel.

pub fn header(&self) -> &Header

Returns a reference to the Header associated with this channel.

Trait Implementations

impl Debug for ShmWriter

fn fmt(&self, f: &mut Formatter) -> Result<(), Error>

Formats the value using the given formatter.

impl Drop for ShmWriter

fn drop(&mut self)

Marks this channel as closed, flushes the changes to the disk, and removes the memory mapping.

impl Writer for ShmWriter

fn write(&mut self, data: &[u8], len: u32) -> Result<u32, WriteError>

Writes a message into the channel. This operation will copy the message into the channel storage. While this is a non blocking operation, only one write should be executed at any given time.

Returns the total amount of bytes wrote into the channel which includes, the size of the message, the size of the message header and the amount of padding add to that message.

Arguments

*data - The buffer which contains the data which is going to be wrote into the channel.

• len - The amount of data which is going to be wrote into to he channel

Errors

Two types of failures may occur: message size is larger than the maximum allowed, or the there is not enough space in the channel to write that message. In the second case, a future write may succeed, if the message has a smaller size that the current one.

use kekbit_core::tick::TickUnit::Nanos;
use kekbit_core::shm::*;
use kekbit_core::header::Header;
use kekbit_core::api::Writer;

const FOREVER: u64 = 99_999_999_999;
let writer_id = 1850;
let channel_id = 42;
let capacity = 30_000;
let max_msg_len = 100;
let header = Header::new(writer_id, channel_id, capacity, max_msg_len, FOREVER, Nanos);
let test_tmp_dir = tempdir::TempDir::new("kektest").unwrap();
let mut writer = shm_writer(&test_tmp_dir.path(), &header).unwrap();
let msg = "There are 10 kinds of people: those who know binary and those who don't";
let msg_data = msg.as_bytes();
writer.write(&msg_data, msg_data.len() as u32).unwrap();

fn flush(&mut self) -> Result<(), Error>

Flushes the channel's outstanding memory map modifications to disk. Calling this method explicitly it is not encouraged as flushing does occur automatically and comes with a performance penalty. It should be used only if for various reasons a writer wants to persist the channel data to the disk at a higher rate than is done automatically.

Returns Ok(()) if the operation succeeds.

Errors

If flushing fails an I/O error is returned.

Examples

use kekbit_core::tick::TickUnit::Nanos;
use kekbit_core::shm::*;
use kekbit_core::header::Header;
use kekbit_core::api::Writer;

const FOREVER: u64 = 99_999_999_999;
let writer_id = 1850;
let channel_id = 42;
let capacity = 30_000;
let max_msg_len = 100;
let header = Header::new(writer_id, channel_id, capacity, max_msg_len, FOREVER, Nanos);
let test_tmp_dir = tempdir::TempDir::new("kektest").unwrap();
let mut writer = shm_writer(&test_tmp_dir.path(), &header).unwrap();
let msg = "There are 10 kinds of people: those who know binary and those who don't";
let msg_data = msg.as_bytes();
writer.write(&msg_data, msg_data.len() as u32).unwrap();
writer.flush().unwrap();

fn heartbeat(&mut self) -> Result<u32, WriteError>

Writes into the stream a heartbeat message. This method shall be used by all writers which want to respect to timeout interval associated to a channel. Hearbeating is the expected mechanism by which a channel writer will keep the active readers interested in the data published on the channel. Heartbeat shall be done regularly, at a time interval which ensures that at least one heartbeat is sent between any two 'timeout' long intervals.