Module dispatch 

One endpoint’s requests, routed to any number of concurrent calls (story C-4).

An endpoint hands out exactly one Receiver<Incoming>. Every request that arrives, for any call in any dialog, comes out of that one stream — so an application holding more than one call had to write its own demultiplexer, and serve dropped whatever the single call it drove did not claim. Both are ways to lose an ACK, which is the loss that leaks calls: nothing retransmits it once Timer H expires and no timer reaps the dialog it would have completed.

This is that demultiplexer, written once. It owns the receiver, routes each request to the call it belongs to, and gives every request that belongs to no call a defined answer instead of silence. The decision table, the counters and the vectors the tests are derived from are in docs/specs/call-dispatch.md.

use std::net::IpAddr;
use sipx_call::{Dispatched, Dispatcher, serve};
use tokio::task::JoinSet;

const MAX_CALLS: usize = 64;
let media_address: IpAddr = "203.0.113.7".parse()?;
let mut dispatcher = Dispatcher::new(endpoint.clone(), incoming);
let mut calls = JoinSet::new();
let outcome: Result<(), Box<dyn std::error::Error>> = async {
    loop {
        tokio::select! {
            event = dispatcher.next() => {
                let Some(event) = event else { break };
                if let Dispatched::Invitation(invitation) = event {
                    if calls.len() >= MAX_CALLS {
                        invitation.refuse(&endpoint, 503, "Service Unavailable").await?;
                        continue;
                    }
                    let mut call = invitation.answer(&endpoint, media_address).await?;
                    let (_, mut requests) = invitation.into_parts();
                    calls.spawn(async move { serve(&mut call, &mut requests).await });
                }
            }
            joined = calls.join_next(), if !calls.is_empty() => {
                if let Some(joined) = joined { joined??; }
            }
        }
    }
    Ok(())
}.await;
calls.shutdown().await;
outcome

Structs

Calls

The set of calls a dispatcher routes to: a cheap, cloneable handle to its routing table.

DispatchCounts

What a dispatcher has refused, shed or could not place.

Dispatcher

One endpoint’s incoming requests, routed to any number of concurrent calls.

DrainProgress

Work observed while a dispatcher is draining.

DrainReport

Terminal result of Dispatcher::drain.

Invitation

An incoming call: the INVITE, and the inbox of the call it may become.

Enums

Dispatched

What the dispatcher could not place itself, handed to the application.

Constants

DEFAULT_QUEUE

How many requests one call’s inbox holds before the dispatcher sheds for it.