Function subscribe_waits_for_the_stream 

pub async fn subscribe_waits_for_the_stream<'a, U, D, DFut, R, RFut, F, FFut>(
    ud: &'a U,
    rx: &mut Receiver<Envelope>,
    source: &str,
    id: &InstrumentId,
    price: Price,
    drive: D,
    release: R,
    fill: F,
)
where
    U: OrderExecutor + ExecutionSource + ?Sized,
    D: FnMut(SubscribeCall<'a>) -> DFut,
    DFut: Future<Output = Result<SubscriptionId, Error>> + 'a,
    R: FnOnce() -> RFut,
    RFut: Future<Output = ()>,
    F: FnOnce(ClientOrderId) -> FFut,
    FFut: Future<Output = ()>,

subscribe_order_state and subscribe_executions return when the stream is open, not before.

A bot places an order right after subscribe_* returns. If the call returned before the broker had opened the stream, an order that is filled in that window would never be reported: the event happened while nobody was listening. So a call returns Ok only when the stream is live, and a stream that does not open in time is an error, with nothing left half open.

The adapter holds the opening of the streams at the broker before the call (the mock’s stream_gate; for a connector that opens its sockets at start, before the connector is started). Then:

1. drive runs the subscribe_order_state future to its end while the broker holds the stream and the clock moves past the connector’s wait (brokapi_mocks::time::elapse_when). The outcome is [Error::Transport], not Ok: the stream is not open.
2. The same for subscribe_executions.
3. release lets the broker open the streams.
4. Both calls are repeated and succeed: the failed calls left nothing behind that blocks them.
5. An order is placed and then filled by fill: the Order event arrives on the order state subscription, and the Execution event on the executions subscription. Both came after the calls returned.

The command gate must be open, so call reconcile first. The order is bought far from the market, two lots.

Panics

With [conformance:subscribe_waits_for_the_stream] <reason>.

Examples

// The broker is slow to open its sockets: held before the connector starts, as BCS opens them at start.
opts.stream_gate.hold();
brokapi_conformance::reconcile(&ud).await;
brokapi_conformance::subscribe_waits_for_the_stream(
    &ud,
    &mut rx,
    "bcs-ud",
    &sber,
    Price::from(100),
    |call| {
        brokapi_mocks::time::elapse_when(
            "the socket request reached the broker",
            || opts.stream_gate.waiting() > 0,
            Duration::from_secs(10),
            call,
        )
    },
    || async { opts.stream_gate.release() },
    |cid| {
        let _ = server.fill(&cid.hyphenated().to_string(), Qty::from(20), Price::from(100));
        async {}
    },
)
.await;