rig_core/observe/
adapter.rs

1//! Execution-local correlation and facts from provider request boundaries.
2//!
3//! ```
4//! use rig_core::observe::scrub_diagnostic;
5//!
6//! assert_eq!(scrub_diagnostic("Bearer secret", &[]), "[redacted]");
7//! ```
8
9use std::sync::{Arc, Mutex};
10
11use serde::{Deserialize, Serialize};
12
13use super::{Action, Emitter, Observation, Stage, Subject, Witness};
14
15/// Bounds diagnostic text and removes control characters. Known credentials
16/// or credential markers redact the whole message; oversized messages are
17/// replaced rather than truncated to a potentially sensitive prefix.
18pub fn scrub_diagnostic(value: &str, secrets: &[String]) -> String {
19    scrub::text(value, secrets)
20}
21
22/// The secrets a URL carries for diagnostic redaction: userinfo and known
23/// credential query parameters, origin-form request URIs included. The
24/// returned values are secrets: keep them runtime-only and never write
25/// them into a diagnostic or an artifact.
26pub fn diagnostic_url_secrets(url: &str) -> Vec<String> {
27    scrub::url_secrets(url)
28}
29
30/// One fact about a provider operation, independently of its effect record.
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32pub struct AdapterObservation {
33    /// Caller-owned logical operation identity within this execution.
34    pub operation: String,
35    /// One-based HTTP send ordinal; absent only when correlation is exhausted.
36    pub attempt: Option<u64>,
37    /// One-based host dispatch attempt, when explicitly supplied by the host.
38    /// Independent of the HTTP send ordinal: a dispatch may send more than once.
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub host_attempt: Option<std::num::NonZeroU64>,
41    /// What the owning boundary observed.
42    pub event: AdapterEvent,
43    /// Scrubbed volatile diagnostics, excluded from semantic comparison.
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub analysis: Option<AdapterAnalysis>,
46}
47
48/// Metadata from the provider boundary. Request/response bodies are not included.
49#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
50#[serde(tag = "event", rename_all = "snake_case")]
51pub enum AdapterEvent {
52    /// Fields actually present in a provider payload; absent fields are not updates.
53    Provider {
54        /// The provider's verdict and selected model, independent of transport closure.
55        verdict: AdapterVerdict,
56    },
57    /// A provider error envelope, including envelopes carried under HTTP 200.
58    ErrorEnvelope {
59        /// Scrubbed original envelope fields, not an inferred HTTP status.
60        error: AdapterErrorEnvelope,
61    },
62    /// The attempt's normalized usage, observed once when its reply ends.
63    Usage {
64        /// The counters of the response's usage.
65        usage: AdapterUsage,
66    },
67    /// A request is about to be sent.
68    Started {
69        /// HTTP method.
70        method: String,
71        /// Provider-declared route template, never a credential-bearing URI.
72        route: String,
73    },
74    /// The transport returned response headers.
75    Response {
76        /// Actual HTTP response status, including successful responses.
77        status: u16,
78    },
79    /// The body reached EOF, independently of any provider terminal verdict.
80    TransportEof {
81        /// Number of complete frames, excluding provider-recognized analysis-only frames.
82        after: usize,
83        /// Undelimited SSE bytes remaining at EOF; zero means no partial frame.
84        partial_bytes: usize,
85    },
86    /// The owning request boundary closed this attempt.
87    Finished {
88        /// Whether the attempt completed, failed, or was dropped.
89        ending: AdapterEnding,
90    },
91    /// A complete transport frame could not be decoded; the driver may continue.
92    Corrupt {
93        /// One-based ordinal counting known, unknown and corrupt frames,
94        /// excluding provider-recognized analysis-only frames.
95        frame: usize,
96    },
97    /// The context cannot assign another unique send ordinal.
98    IdentityExhausted,
99}
100
101/// Sparse provider metadata: a present field supersedes its previous value.
102#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
103pub struct AdapterVerdict {
104    /// Provider finish code, including unknown codes after scrubbing.
105    pub finish_reason: Option<String>,
106    /// Provider prompt-block/refusal reason, when reported.
107    pub block_reason: Option<String>,
108    /// Provider detail accompanying a finish/block reason.
109    pub detail: Option<String>,
110    /// Actual provider model/version, not the requested alias.
111    pub model: Option<String>,
112}
113
114/// Analysis-only response metadata. Never use identifiers as operation join keys.
115#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
116pub struct AdapterAnalysis {
117    /// Provider response identifier, scrubbed and bounded.
118    pub response_id: Option<String>,
119    /// Allowlisted response headers. None means the transport supplied no map;
120    /// an empty map means it supplied no allowlisted values.
121    pub headers: Option<std::collections::BTreeMap<String, String>>,
122}
123
124/// Original provider error-envelope fields, separate from Rig retry classification.
125#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
126pub struct AdapterErrorEnvelope {
127    /// Provider code in its scalar wire spelling; unknown codes stay observable.
128    pub code: Option<String>,
129    /// Provider status name, when present.
130    pub status: Option<String>,
131    /// Bounded, scrubbed provider message, when present.
132    pub message: Option<String>,
133}
134
135/// Error-envelope projection with optional code, message, and `type`/`status`.
136/// Codes remain JSON values until emission validates their scalar shape.
137#[derive(Default)]
138pub struct ObservedError {
139    pub code: Option<serde_json::Value>,
140    pub kind: Option<String>,
141    pub message: Option<String>,
142}
143
144impl ObservedError {
145    /// The error envelope a provider document carries, located as its
146    /// [`crate::provider_response::ProviderResponseError`] reads it: an
147    /// error event's own `type` is its tag, not a kind, and an `error`
148    /// string is the message alone.
149    pub fn of(doc: &serde_json::Value) -> Option<Self> {
150        let (error, nested) = crate::provider_response::located_error(doc)?;
151        let text = |key: &str| error.get(key).and_then(serde_json::Value::as_str);
152        let kind = nested.then(|| text("type").or_else(|| text("status")));
153        Some(Self {
154            code: error.get("code").filter(|code| !code.is_null()).cloned(),
155            kind: kind.flatten().map(str::to_owned),
156            message: error
157                .as_str()
158                .or_else(|| text("message"))
159                .map(str::to_owned),
160        })
161    }
162
163    /// Emit this envelope as the attempt's error-envelope fact, scrubbed.
164    pub fn emit(self, sink: &mut ObservationSink<'_>) {
165        let code = self.code.map(|code| match code {
166            serde_json::Value::String(code) => sink.scrub(&code),
167            serde_json::Value::Number(code) => code.to_string(),
168            _ => "[invalid]".to_owned(),
169        });
170        sink.error(AdapterErrorEnvelope {
171            code,
172            status: self.kind.map(|value| sink.scrub(&value)),
173            message: self.message.map(|value| sink.scrub(&value)),
174        });
175    }
176}
177
178/// The counters of the [`Usage`](crate::completion::Usage) the decoder
179/// normalized into the reply's end, observed once when the reply ends; an
180/// attempt that fails before its end observes none. Unreported is `None`.
181#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
182pub struct AdapterUsage {
183    /// Every input token: uncached, cached, cache writes and hosted-tool prompt.
184    pub input_tokens: Option<u64>,
185    /// Every output token, reasoning included.
186    pub output_tokens: Option<u64>,
187    /// Always `None`; kept for serde compatibility. Read the response's
188    /// [`Usage::total_tokens`](crate::completion::Usage::total_tokens).
189    pub total_tokens: Option<u64>,
190    /// The part of the input read from a provider-managed cache.
191    pub cached_input_tokens: Option<u64>,
192    /// The part of the output spent on internal reasoning.
193    pub reasoning_tokens: Option<u64>,
194    /// The part of the input a provider's hosted tools added to the prompt.
195    pub tool_input_tokens: Option<u64>,
196}
197
198impl From<&crate::completion::Usage> for AdapterUsage {
199    fn from(usage: &crate::completion::Usage) -> Self {
200        // Exhaustive: a new `Usage` counter must be mapped or ignored here.
201        let crate::completion::Usage {
202            input_tokens,
203            output_tokens,
204            total_tokens: _,
205            cached_input_tokens,
206            cache_creation_input_tokens: _,
207            tool_use_prompt_tokens,
208            reasoning_tokens,
209            cost: _,
210        } = *usage;
211        Self {
212            input_tokens,
213            output_tokens,
214            total_tokens: None,
215            cached_input_tokens,
216            reasoning_tokens,
217            tool_input_tokens: tool_use_prompt_tokens,
218        }
219    }
220}
221
222/// Boundary identified by the adapter's original typed error, not its message.
223#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
224#[serde(rename_all = "snake_case")]
225pub enum AdapterErrorBoundary {
226    /// Request construction or validation failed.
227    Request,
228    /// The provider returned a failure response or error envelope.
229    ProviderResponse,
230    /// Response decoding or validation failed.
231    Decode,
232    /// A typed transport termination was reported.
233    Transport,
234    /// An erased client error does not establish a more specific boundary.
235    #[default]
236    Unknown,
237}
238
239impl AdapterErrorBoundary {
240    pub(crate) fn from_http(error: &crate::http_client::Error) -> Self {
241        use crate::http_client::Error as H;
242        match error {
243            H::Protocol(_) | H::InvalidHeaderValue(_) | H::NoHeaders | H::Unbuilt(_) => {
244                Self::Request
245            }
246            H::InvalidContentType(_) => Self::Decode,
247            H::StreamEnded => Self::Transport,
248            H::InvalidStatusCodeWithDetails { .. } => Self::ProviderResponse,
249            _ => Self::Unknown,
250        }
251    }
252}
253
254/// How a provider attempt closed, distinct from the run's ending.
255#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
256#[serde(tag = "ending", rename_all = "snake_case")]
257pub enum AdapterEnding {
258    /// The request produced a decoded response.
259    Decoded,
260    /// The request failed; provider payloads remain in the existing error report.
261    Error {
262        /// Known boundary before error conversion erases transport subtypes.
263        #[serde(default)]
264        boundary: AdapterErrorBoundary,
265        /// Stable Rig error classification.
266        kind: String,
267        /// HTTP status reported by the failure, when available.
268        status: Option<u16>,
269        /// Retryability under the existing policy, not a decision to retry.
270        retryable: bool,
271    },
272    /// A provider terminal record was decoded from the stream.
273    Terminal,
274    /// Transport EOF without a provider terminal record.
275    Eof {
276        /// Number of complete frames, excluding provider-recognized analysis-only frames.
277        after: usize,
278    },
279    /// EOF left an undelimited SSE event; bytes are never retained in the fact.
280    PartialFrame {
281        /// Raw bytes since the last blank event delimiter, saturated at usize::MAX.
282        byte_count: usize,
283        /// Complete frames previously delivered, excluding analysis-only frames.
284        after: usize,
285    },
286    /// The future or stream was dropped before its boundary closed.
287    Dropped,
288}
289
290/// A caller-supplied witness and logical operation identity.
291///
292/// Clone this context for retries of the same operation. Create a new context
293/// with a distinct identity for a different call, including identical parallel
294/// requests. Identity is execution-local; it does not assert replay equivalence.
295/// This handle is runtime-only and must never be serialized into provider data.
296#[derive(Clone)]
297pub struct AdapterContext {
298    inner: Arc<AdapterContextInner>,
299}
300
301struct AdapterContextInner {
302    sink: Arc<dyn Witness + Send + Sync>,
303    subject: Subject,
304    operation: String,
305    next: Arc<Mutex<Option<u64>>>,
306    host_attempt: Option<std::num::NonZeroU64>,
307}
308
309impl std::fmt::Debug for AdapterContext {
310    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
311        // Caller-supplied identity can contain sensitive data; do not log it.
312        f.debug_struct("AdapterContext").finish_non_exhaustive()
313    }
314}
315
316impl AdapterContext {
317    /// Caller-owned identity of this logical operation within the execution.
318    pub fn operation(&self) -> &str {
319        &self.inner.operation
320    }
321
322    /// Bind a logical operation to its witness and existing subject.
323    /// The caller must use a non-sensitive, execution-unique operation identity.
324    pub fn new(
325        sink: Arc<dyn Witness + Send + Sync>,
326        subject: Subject,
327        operation: impl Into<String>,
328    ) -> Self {
329        Self {
330            inner: Arc::new(AdapterContextInner {
331                sink,
332                subject,
333                operation: operation.into(),
334                next: Arc::new(Mutex::new(Some(1))),
335                host_attempt: None,
336            }),
337        }
338    }
339
340    /// Bind an explicit host attempt to its current dispatch subject.
341    ///
342    /// The logical operation, witness and HTTP send counter remain shared.
343    /// Existing clones retain their own subject and host ordinal, so an older
344    /// in-flight attempt cannot be relabeled by a concurrent retry. The host
345    /// owns ordinal allocation and must start a new context for a changed
346    /// logical request rather than reusing this operation identity.
347    pub fn for_host_attempt(&self, subject: Subject, attempt: std::num::NonZeroU64) -> Self {
348        Self {
349            inner: Arc::new(AdapterContextInner {
350                sink: self.inner.sink.clone(),
351                subject,
352                operation: self.inner.operation.clone(),
353                next: self.inner.next.clone(),
354                host_attempt: Some(attempt),
355            }),
356        }
357    }
358
359    /// Begins a send and captures request credentials for diagnostic scrubbing.
360    /// `route` must be a credential-free provider template without base-URL
361    /// prefixes or query data. Returns `None` when attempt IDs are exhausted.
362    pub(crate) fn attempt_for<B>(
363        &self,
364        request: &http::Request<B>,
365        route: &str,
366    ) -> Option<AdapterAttempt> {
367        let mut attempt = self.begin(request.method(), route)?;
368        attempt.secrets = scrub::request_secrets(request);
369        Some(attempt)
370    }
371
372    fn emit(&self, attempt: Option<u64>, event: AdapterEvent) {
373        self.emit_with_analysis(attempt, event, None);
374    }
375
376    fn emit_with_analysis(
377        &self,
378        attempt: Option<u64>,
379        event: AdapterEvent,
380        analysis: Option<AdapterAnalysis>,
381    ) {
382        self.inner.sink.observe(Observation::new(
383            self.inner.subject.clone(),
384            Stage::Handler,
385            Emitter::named("rig-core/adapter"),
386            Action::Adapter {
387                observation: AdapterObservation {
388                    operation: self.inner.operation.clone(),
389                    attempt,
390                    host_attempt: self.inner.host_attempt,
391                    event,
392                    analysis,
393                },
394            },
395        ));
396    }
397
398    /// Allocates a send ordinal and emits its start under a credential-free
399    /// route template. Emits exhaustion at the last ordinal; later calls return
400    /// `None` without reusing identities or affecting provider execution.
401    pub(crate) fn begin(&self, method: &http::Method, route: &str) -> Option<AdapterAttempt> {
402        let number = {
403            let mut next = self
404                .inner
405                .next
406                .lock()
407                .unwrap_or_else(std::sync::PoisonError::into_inner);
408            let number = (*next)?;
409            *next = number.checked_add(1);
410            number
411        };
412        self.emit(
413            Some(number),
414            AdapterEvent::Started {
415                method: method.to_string(),
416                route: route.to_owned(),
417            },
418        );
419        if number == u64::MAX {
420            self.emit(None, AdapterEvent::IdentityExhausted);
421        }
422        Some(AdapterAttempt {
423            context: self.clone(),
424            number,
425            closed: false,
426            response_seen: false,
427            unframed: 0,
428            secrets: Vec::new(),
429            pending_response_id: None,
430            error_boundary: None,
431        })
432    }
433}
434
435/// The request owns this guard until completion or cancellation.
436pub(crate) struct AdapterAttempt {
437    context: AdapterContext,
438    number: u64,
439    closed: bool,
440    response_seen: bool,
441    /// Bytes the transport's framer had not framed when the body ended.
442    unframed: usize,
443    secrets: Vec<String>,
444    pending_response_id: Option<String>,
445    error_boundary: Option<AdapterErrorBoundary>,
446}
447
448impl AdapterAttempt {
449    pub(crate) fn text(&self, text: &str) -> String {
450        scrub::text(text, &self.secrets)
451    }
452
453    fn emit_with_analysis(&self, event: AdapterEvent, analysis: AdapterAnalysis) {
454        let analysis = (analysis != AdapterAnalysis::default()).then_some(analysis);
455        self.context
456            .emit_with_analysis(Some(self.number), event, analysis);
457    }
458
459    fn emit(&self, event: AdapterEvent) {
460        self.context.emit(Some(self.number), event);
461    }
462
463    /// Project a payload's facts through the projector that understands it.
464    pub(crate) fn project(&mut self, project: impl FnOnce(&mut ObservationSink<'_>)) {
465        project(&mut ObservationSink { attempt: self });
466    }
467
468    pub(crate) fn provider(&mut self, verdict: AdapterVerdict, response_id: Option<String>) {
469        if response_id.is_some() {
470            self.pending_response_id = response_id;
471        }
472        // An ID-only payload must not manufacture a semantic provider event.
473        // Retain at most one ID until the next verdict or the attempt closure.
474        if verdict != AdapterVerdict::default() {
475            let response_id = self.pending_response_id.take();
476            self.emit_with_analysis(
477                AdapterEvent::Provider { verdict },
478                AdapterAnalysis {
479                    response_id,
480                    ..AdapterAnalysis::default()
481                },
482            );
483        }
484    }
485
486    pub(crate) fn response_with_headers(
487        &mut self,
488        status: http::StatusCode,
489        headers: Option<&http::HeaderMap>,
490    ) {
491        if !self.response_seen {
492            self.response_seen = true;
493            self.emit_with_analysis(
494                AdapterEvent::Response {
495                    status: status.as_u16(),
496                },
497                AdapterAnalysis {
498                    headers: headers.map(|h| scrub::headers(h, &self.secrets)),
499                    ..AdapterAnalysis::default()
500                },
501            );
502        }
503    }
504
505    pub(crate) fn finish(&mut self, ending: AdapterEnding) {
506        if !self.closed {
507            self.closed = true;
508            let response_id = self.pending_response_id.take();
509            self.emit_with_analysis(
510                AdapterEvent::Finished { ending },
511                AdapterAnalysis {
512                    response_id,
513                    ..AdapterAnalysis::default()
514                },
515            );
516        }
517    }
518}
519
520impl Drop for AdapterAttempt {
521    fn drop(&mut self) {
522        self.finish(AdapterEnding::Dropped);
523    }
524}
525
526/// Where a payload projector writes its observation facts.
527///
528/// A projector reads verdicts, ids and error envelopes off a raw payload
529/// before normalization discards them. Text it forwards must go through
530/// [`Self::scrub`]: a payload can echo credentials.
531///
532/// Usage is not a projector's to report: the driver observes it once, from
533/// the usage the decoder normalized into the reply's end.
534///
535/// ```compile_fail
536/// use rig_core::wire::{AdapterEvent, AdapterUsage, ObservationSink};
537///
538/// // A projector cannot parse usage a second time off the raw payload.
539/// fn project(sink: &mut ObservationSink<'_>) {
540///     sink.emit(AdapterEvent::Usage {
541///         usage: AdapterUsage::default(),
542///     });
543/// }
544/// ```
545pub struct ObservationSink<'a> {
546    attempt: &'a mut AdapterAttempt,
547}
548
549impl ObservationSink<'_> {
550    /// Record the provider's error envelope.
551    pub fn error(&mut self, error: AdapterErrorEnvelope) {
552        self.attempt.emit(AdapterEvent::ErrorEnvelope { error });
553    }
554
555    /// Record the provider's verdict, and the response id it named.
556    pub fn provider(&mut self, verdict: AdapterVerdict, response_id: Option<String>) {
557        self.attempt.provider(verdict, response_id);
558    }
559
560    /// Bound and redact diagnostic text from the payload.
561    pub fn scrub(&self, value: &str) -> String {
562        self.attempt.text(value)
563    }
564}
565
566/// Shared by the SSE transport and frame driver, which own different boundaries.
567#[derive(Clone, Default)]
568pub(crate) struct AdapterSlot(Arc<Mutex<Option<AdapterAttempt>>>);
569
570impl AdapterSlot {
571    pub(crate) fn transport_eof(&self, after: usize) {
572        if let Some(attempt) = self
573            .0
574            .lock()
575            .unwrap_or_else(std::sync::PoisonError::into_inner)
576            .as_ref()
577        {
578            attempt.emit(AdapterEvent::TransportEof {
579                after,
580                partial_bytes: attempt.unframed,
581            });
582        }
583    }
584
585    /// Project a reply payload's facts through the projector that
586    /// understands it.
587    pub(crate) fn project(&self, project: impl FnOnce(&mut ObservationSink<'_>)) {
588        if let Some(attempt) = self
589            .0
590            .lock()
591            .unwrap_or_else(std::sync::PoisonError::into_inner)
592            .as_mut()
593        {
594            attempt.project(project);
595        }
596    }
597
598    /// Close the attempt as its reply ended: the one place usage is
599    /// observed, once, from the decoder's end, just before `Terminal`.
600    pub(crate) fn terminal(&self, usage: Option<&crate::completion::Usage>) {
601        let mut attempt = self
602            .0
603            .lock()
604            .unwrap_or_else(std::sync::PoisonError::into_inner);
605        if let Some(attempt) = attempt.as_mut().filter(|attempt| !attempt.closed) {
606            let usage = usage.map(AdapterUsage::from);
607            if let Some(usage) = usage.filter(|usage| *usage != AdapterUsage::default()) {
608                attempt.emit(AdapterEvent::Usage { usage });
609            }
610            attempt.finish(AdapterEnding::Terminal);
611        }
612    }
613
614    /// Install the attempt this send's facts belong to.
615    pub(crate) fn install(&self, attempt: Option<AdapterAttempt>) {
616        *self
617            .0
618            .lock()
619            .unwrap_or_else(std::sync::PoisonError::into_inner) = attempt;
620    }
621
622    pub(crate) fn response(&self, status: http::StatusCode) {
623        self.response_with_headers(status, None);
624    }
625
626    pub(crate) fn response_with_headers(
627        &self,
628        status: http::StatusCode,
629        headers: Option<&http::HeaderMap>,
630    ) {
631        if let Some(attempt) = self
632            .0
633            .lock()
634            .unwrap_or_else(std::sync::PoisonError::into_inner)
635            .as_mut()
636        {
637            attempt.response_with_headers(status, headers);
638        }
639    }
640
641    fn finish(&self, ending: AdapterEnding) {
642        if let Some(attempt) = self
643            .0
644            .lock()
645            .unwrap_or_else(std::sync::PoisonError::into_inner)
646            .as_mut()
647        {
648            attempt.finish(ending);
649        }
650    }
651
652    /// Preserve information before the SSE transport converts the owned error.
653    pub(crate) fn error_boundary(&self, boundary: AdapterErrorBoundary) {
654        if let Some(attempt) = self
655            .0
656            .lock()
657            .unwrap_or_else(std::sync::PoisonError::into_inner)
658            .as_mut()
659        {
660            attempt.error_boundary = Some(boundary);
661        }
662    }
663
664    pub(crate) fn fail(&self, error: &crate::error::ProviderError) {
665        if let Some(status) = error.provider_response_status() {
666            self.response(status);
667        }
668        let report = error.report();
669        let boundary = self
670            .0
671            .lock()
672            .unwrap_or_else(std::sync::PoisonError::into_inner)
673            .as_ref()
674            .and_then(|attempt| attempt.error_boundary)
675            .unwrap_or_else(|| error.boundary());
676        self.finish(AdapterEnding::Error {
677            boundary,
678            kind: report.kind.code().to_owned(),
679            status: report.http_status,
680            retryable: report.is_retryable(),
681        });
682    }
683
684    /// Record the bytes the transport's framer had not framed when the
685    /// reply's body ended.
686    pub(crate) fn unframed(&self, bytes: usize) {
687        if let Some(attempt) = self
688            .0
689            .lock()
690            .unwrap_or_else(std::sync::PoisonError::into_inner)
691            .as_mut()
692        {
693            attempt.unframed = bytes;
694        }
695    }
696
697    pub(crate) fn eof(&self, after: usize) {
698        if let Some(attempt) = self
699            .0
700            .lock()
701            .unwrap_or_else(std::sync::PoisonError::into_inner)
702            .as_mut()
703        {
704            let byte_count = attempt.unframed;
705            let ending = if byte_count == 0 {
706                AdapterEnding::Eof { after }
707            } else {
708                AdapterEnding::PartialFrame { byte_count, after }
709            };
710            attempt.finish(ending);
711        }
712    }
713
714    pub(crate) fn corrupt(&self, frame: usize) {
715        if let Some(attempt) = self
716            .0
717            .lock()
718            .unwrap_or_else(std::sync::PoisonError::into_inner)
719            .as_ref()
720        {
721            attempt
722                .context
723                .emit(Some(attempt.number), AdapterEvent::Corrupt { frame });
724        }
725    }
726}
727
728mod scrub;
729#[cfg(test)]
730mod tests;
rig_core/observe/adapter.rs

rig_core/observe/
adapter.rs
