Struct fuel_core_interfaces::txpool::Sender

pub struct Sender(_);

RPC client for doing calls to the TxPool through an MPSC channel.

Implementations

impl Sender

pub fn new(sender: Sender<TxPoolMpsc>) -> Self

pub async fn insert(
    &self,
    txs: Vec<Arc<Transaction>>
) -> Result<Vec<Result<InsertionResult>>>

pub async fn find(&self, ids: Vec<TxId>) -> Result<Vec<Option<TxInfo>>>

pub async fn find_one(&self, id: TxId) -> Result<Option<TxInfo>>

pub async fn find_dependent(&self, ids: Vec<TxId>) -> Result<Vec<ArcPoolTx>>

pub async fn filter_by_negative(&self, ids: Vec<TxId>) -> Result<Vec<TxId>>

pub async fn includable(&self) -> Result<Vec<ArcPoolTx>>

pub async fn remove(&self, ids: Vec<TxId>) -> Result<Vec<ArcPoolTx>>

pub fn channel(buffer: usize) -> (Sender, Receiver<TxPoolMpsc>)

Methods from Deref<Target = Sender<TxPoolMpsc>>

pub async fn send(
    &self,
    value: T
) -> impl Future<Output = Result<(), SendError<T>>>

Sends a value, waiting until there is capacity.

A successful send occurs when it is determined that the other end of the channel has not hung up already. An unsuccessful send would be one where the corresponding receiver has already been closed. Note that a return value of Err means that the data will never be received, but a return value of Ok does not mean that the data will be received. It is possible for the corresponding receiver to hang up immediately after this function returns Ok.

Errors

If the receive half of the channel is closed, either due to close being called or the Receiver handle dropping, the function returns an error. The error includes the value passed to send.

Cancel safety

If send is used as the event in a tokio::select! statement and some other branch completes first, then it is guaranteed that the message was not sent.

This channel uses a queue to ensure that calls to send and reserve complete in the order they were requested. Cancelling a call to send makes you lose your place in the queue.

Examples

In the following example, each call to send will block until the previously sent value was received.

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    tokio::spawn(async move {
        for i in 0..10 {
            if let Err(_) = tx.send(i).await {
                println!("receiver dropped");
                return;
            }
        }
    });

    while let Some(i) = rx.recv().await {
        println!("got = {}", i);
    }
}

pub async fn closed(&self) -> impl Future<Output = ()>

Completes when the receiver has dropped.

This allows the producers to get notified when interest in the produced values is canceled and immediately stop doing work.

Cancel safety

This method is cancel safe. Once the channel is closed, it stays closed forever and all future calls to closed will return immediately.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx1, rx) = mpsc::channel::<()>(1);
    let tx2 = tx1.clone();
    let tx3 = tx1.clone();
    let tx4 = tx1.clone();
    let tx5 = tx1.clone();
    tokio::spawn(async move {
        drop(rx);
    });

    futures::join!(
        tx1.closed(),
        tx2.closed(),
        tx3.closed(),
        tx4.closed(),
        tx5.closed()
    );
    println!("Receiver dropped");
}

pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>>

Attempts to immediately send a message on this Sender

This method differs from send by returning immediately if the channel’s buffer is full or no receiver is waiting to acquire some data. Compared with send, this function has two failure cases instead of one (one for disconnection, one for a full buffer).

Errors

If the channel capacity has been reached, i.e., the channel has n buffered values where n is the argument passed to channel, then an error is returned.

If the receive half of the channel is closed, either due to close being called or the Receiver handle dropping, the function returns an error. The error includes the value passed to send.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // Create a channel with buffer size 1
    let (tx1, mut rx) = mpsc::channel(1);
    let tx2 = tx1.clone();

    tokio::spawn(async move {
        tx1.send(1).await.unwrap();
        tx1.send(2).await.unwrap();
        // task waits until the receiver receives a value.
    });

    tokio::spawn(async move {
        // This will return an error and send
        // no message if the buffer is full
        let _ = tx2.try_send(3);
    });

    let mut msg;
    msg = rx.recv().await.unwrap();
    println!("message {} received", msg);

    msg = rx.recv().await.unwrap();
    println!("message {} received", msg);

    // Third message may have never been sent
    match rx.recv().await {
        Some(msg) => println!("message {} received", msg),
        None => println!("the third message was never sent"),
    }
}

pub async fn send_timeout(
    &self,
    value: T,
    timeout: Duration
) -> impl Future<Output = Result<(), SendTimeoutError<T>>>

Sends a value, waiting until there is capacity, but only for a limited time.

Shares the same success and error conditions as send, adding one more condition for an unsuccessful send, which is when the provided timeout has elapsed, and there is no capacity available.

Errors

If the receive half of the channel is closed, either due to close being called or the Receiver having been dropped, the function returns an error. The error includes the value passed to send.

Panics

This function panics if it is called outside the context of a Tokio runtime with time enabled.

Examples

In the following example, each call to send_timeout will block until the previously sent value was received, unless the timeout has elapsed.

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    tokio::spawn(async move {
        for i in 0..10 {
            if let Err(e) = tx.send_timeout(i, Duration::from_millis(100)).await {
                println!("send error: #{:?}", e);
                return;
            }
        }
    });

    while let Some(i) = rx.recv().await {
        println!("got = {}", i);
        sleep(Duration::from_millis(200)).await;
    }
}

pub fn blocking_send(&self, value: T) -> Result<(), SendError<T>>

Blocking send to call outside of asynchronous contexts.

This method is intended for use cases where you are sending from synchronous code to asynchronous code, and will work even if the receiver is not using blocking_recv to receive the message.

Panics

This function panics if called within an asynchronous execution context.

Examples

use std::thread;
use tokio::runtime::Runtime;
use tokio::sync::mpsc;

fn main() {
    let (tx, mut rx) = mpsc::channel::<u8>(1);

    let sync_code = thread::spawn(move || {
        tx.blocking_send(10).unwrap();
    });

    Runtime::new().unwrap().block_on(async move {
        assert_eq!(Some(10), rx.recv().await);
    });
    sync_code.join().unwrap()
}

pub fn is_closed(&self) -> bool

Checks if the channel has been closed. This happens when the Receiver is dropped, or when the Receiver::close method is called.

let (tx, rx) = tokio::sync::mpsc::channel::<()>(42);
assert!(!tx.is_closed());

let tx2 = tx.clone();
assert!(!tx2.is_closed());

drop(rx);
assert!(tx.is_closed());
assert!(tx2.is_closed());

pub async fn reserve(
    &self
) -> impl Future<Output = Result<Permit<'_, T>, SendError<()>>>

Waits for channel capacity. Once capacity to send one message is available, it is reserved for the caller.

If the channel is full, the function waits for the number of unreceived messages to become less than the channel capacity. Capacity to send one message is reserved for the caller. A Permit is returned to track the reserved capacity. The send function on Permit consumes the reserved capacity.

Dropping Permit without sending a message releases the capacity back to the channel.

Cancel safety

This channel uses a queue to ensure that calls to send and reserve complete in the order they were requested. Cancelling a call to reserve makes you lose your place in the queue.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    // Reserve capacity
    let permit = tx.reserve().await.unwrap();

    // Trying to send directly on the `tx` will fail due to no
    // available capacity.
    assert!(tx.try_send(123).is_err());

    // Sending on the permit succeeds
    permit.send(456);

    // The value sent on the permit is received
    assert_eq!(rx.recv().await.unwrap(), 456);
}

pub fn try_reserve(&self) -> Result<Permit<'_, T>, TrySendError<()>>

Tries to acquire a slot in the channel without waiting for the slot to become available.

If the channel is full this function will return TrySendError, otherwise if there is a slot available it will return a Permit that will then allow you to send on the channel with a guaranteed slot. This function is similar to reserve except it does not await for the slot to become available.

Dropping Permit without sending a message releases the capacity back to the channel.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(1);

    // Reserve capacity
    let permit = tx.try_reserve().unwrap();

    // Trying to send directly on the `tx` will fail due to no
    // available capacity.
    assert!(tx.try_send(123).is_err());

    // Trying to reserve an additional slot on the `tx` will
    // fail because there is no capacity.
    assert!(tx.try_reserve().is_err());

    // Sending on the permit succeeds
    permit.send(456);

    // The value sent on the permit is received
    assert_eq!(rx.recv().await.unwrap(), 456);

}

pub fn same_channel(&self, other: &Sender<T>) -> bool

Returns true if senders belong to the same channel.

Examples

let (tx, rx) = tokio::sync::mpsc::channel::<()>(1);
let  tx2 = tx.clone();
assert!(tx.same_channel(&tx2));

let (tx3, rx3) = tokio::sync::mpsc::channel::<()>(1);
assert!(!tx3.same_channel(&tx2));

pub fn capacity(&self) -> usize

Returns the current capacity of the channel.

The capacity goes down when sending a value by calling send or by reserving capacity with reserve. The capacity goes up when values are received by the Receiver. This is distinct from max_capacity, which always returns buffer capacity initially specified when calling channel

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel::<()>(5);

    assert_eq!(tx.capacity(), 5);

    // Making a reservation drops the capacity by one.
    let permit = tx.reserve().await.unwrap();
    assert_eq!(tx.capacity(), 4);

    // Sending and receiving a value increases the capacity by one.
    permit.send(());
    rx.recv().await.unwrap();
    assert_eq!(tx.capacity(), 5);
}

pub fn downgrade(&self) -> WeakSender<T>

Converts the Sender to a WeakSender that does not count towards RAII semantics, i.e. if all Sender instances of the channel were dropped and only WeakSender instances remain, the channel is closed.

pub fn max_capacity(&self) -> usize

Returns the maximum buffer capacity of the channel.

The maximum capacity is the buffer capacity initially specified when calling channel. This is distinct from capacity, which returns the current available buffer capacity: as messages are sent and received, the value returned by capacity will go up or down, whereas the value returned by max_capacity will remain constant.

Examples

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, _rx) = mpsc::channel::<()>(5);
      
    // both max capacity and capacity are the same at first
    assert_eq!(tx.max_capacity(), 5);
    assert_eq!(tx.capacity(), 5);

    // Making a reservation doesn't change the max capacity.
    let permit = tx.reserve().await.unwrap();
    assert_eq!(tx.max_capacity(), 5);
    // but drops the capacity by one
    assert_eq!(tx.capacity(), 4);
}

Trait Implementations

impl Clone for Sender

fn clone(&self) -> Sender

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Deref for Sender

type Target = Sender<TxPoolMpsc>

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl DerefMut for Sender

fn deref_mut(&mut self) -> &mut Self::Target

Mutably dereferences the value.

impl TransactionPool for Sender

fn pending_number<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Result<usize>> + Send + 'async_trait>>where
    Self: 'async_trait,
    'life0: 'async_trait,

Returns the number of pending transactions in the TxPool.

fn total_consumable_gas<'life0, 'async_trait>(
    &'life0 self
) -> Pin<Box<dyn Future<Output = Result<u64>> + Send + 'async_trait>>where
    Self: 'async_trait,
    'life0: 'async_trait,

fn remove_txs<'life0, 'async_trait>(
    &'life0 mut self,
    ids: Vec<TxId>
) -> Pin<Box<dyn Future<Output = Result<Vec<ArcPoolTx>>> + Send + 'async_trait>>where
    Self: 'async_trait,
    'life0: 'async_trait,