Struct ShmWriter

pub struct ShmWriter<D: DataFormat> { /* private fields */ }

An implementation of the Writer which access a persistent channel through memory mapping, and uses a specific DataFormat. 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;
use kekbit_codecs::codecs::raw::RawBinDataFormat;

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, RawBinDataFormat).unwrap();
writer.heartbeat().unwrap();

Implementations

impl<D: DataFormat> ShmWriter<D>

pub fn available(&self) -> u32

Returns the amount of space in this channel still available for write.

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.

pub fn data_format(&self) -> &D

Trait Implementations

impl<D: DataFormat> Drop for ShmWriter<D>

fn drop(&mut self)

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

impl<D: DataFormat> Writer<D> for ShmWriter<D>

fn write(&mut self, data: &impl Encodable<D>) -> Result<u32, WriteError>

Writes a message into the channel. This operation will encode the data directly into channel. 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 data which to encode and write into the channel.

Errors

Two kinds of failures may occur. One if the encoding operation failed, the other if the channel rejected the message for reasons such data is too large or no space is available in the channel.

Examples

use kekbit_core::tick::TickUnit::Nanos;
use kekbit_core::shm::*;
use kekbit_core::header::Header;
use kekbit_core::api::Writer;
use kekbit_codecs::codecs::raw::RawBinDataFormat;

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, RawBinDataFormat).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).unwrap();

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

Push a heartbeat message into the channel. Hearbeats are zero sized messages which do not need encoding. Reader should never activate callbacks for heartbeat messsages.

Returns RecordHeaderLen, 8 in the current version if the operation succeeds.

Errors

If the operation fails a ChannelFull error will be returned, which signals that the channel will not accept any new messages.

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;
use kekbit_codecs::codecs::raw::RawBinDataFormat;

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, RawBinDataFormat).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).unwrap();
writer.flush().unwrap();