aviso 2.2.1

Core client library for aviso-server, ECMWF's notification service.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

// (C) Copyright 2024- ECMWF and individual contributors.
//
// This software is licensed under the terms of the Apache Licence Version 2.0
// which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
// In applying this licence, ECMWF does not waive the privileges and immunities
// granted to it by virtue of its status as an intergovernmental organisation nor
// does it submit to any jurisdiction.

use crate::ClientError;

pub(super) const OPENING_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

pub(super) fn protocol(message: &str) -> ClientError {
    ClientError::StreamProtocol {
        message: format!("{message}; check base_url points to an Aviso server"),
        request_id: None,
    }
}

pub(super) fn validate_content_type(
    headers: &reqwest::header::HeaderMap,
) -> Result<(), ClientError> {
    // Parameters are allowed: `Text/Event-Stream; charset=utf-8` is valid,
    // whereas `text/html` is not an SSE response.
    let valid = headers
        .get(reqwest::header::CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        .is_some_and(|media| media.trim().eq_ignore_ascii_case("text/event-stream"));
    if valid {
        Ok(())
    } else {
        Err(protocol(
            "HTTP 200 response requires Content-Type text/event-stream",
        ))
    }
}

/// Consume only opening frames, leaving all subsequent data for the normal
/// dispatcher. A heartbeat or unknown event cannot confirm an Aviso stream.
pub(super) fn confirmed(
    parser: &mut finesse::Parser,
    historical: bool,
) -> Result<bool, ClientError> {
    while let Some(frame) = parser.next_frame() {
        let finesse::Frame::Message(message) = frame else {
            continue;
        };
        let expected = if historical {
            "replay_started"
        } else {
            "connection_established"
        };
        let expected_event = if historical {
            "replay-control"
        } else {
            "live-notification"
        };
        match message.event.as_str() {
            "live-notification" | "replay" | "replay-control" => {
                let value: serde_json::Value = serde_json::from_str(&message.data)
                    .map_err(|_| protocol("invalid Aviso opening control event"))?;
                if message.event == expected_event
                    && value.get("type").and_then(serde_json::Value::as_str) == Some(expected)
                {
                    return Ok(true);
                }
                return Err(protocol(
                    "unexpected event before required Aviso opening control",
                ));
            }
            "error" => {
                let wire: crate::watch::wire::WireErrorEvent = serde_json::from_str(&message.data)
                    .map_err(|_| protocol("invalid Aviso opening error event"))?;
                return Err(ClientError::StreamProtocol {
                    message: wire.message.or(wire.error).unwrap_or_else(|| {
                        "server rejected stream before opening confirmation".into()
                    }),
                    request_id: wire.request_id,
                });
            }
            "connection-closing" => {
                return Err(protocol("stream closed before opening confirmation"));
            }
            _ => {}
        }
    }
    Ok(false)
}

pub(super) async fn with_startup_budget(
    run: impl std::future::Future<Output = ()>,
    timeout: Option<std::time::Duration>,
    mut ready: tokio::sync::watch::Receiver<bool>,
    errors: tokio::sync::mpsc::Sender<Result<crate::Notification, ClientError>>,
) {
    let Some(timeout) = timeout else {
        run.await;
        return;
    };
    tokio::pin!(run);
    tokio::select! {
        biased;
        () = &mut run => return,
        confirmed = async { ready.wait_for(|confirmed| *confirmed).await.is_ok() } => {
            if !confirmed {
                return;
            }
        }
        () = tokio::time::sleep(timeout) => {
            // No data can precede confirmation, so this channel has room.
            errors.send(Err(ClientError::StreamProtocol {
                message: "listener startup timeout exceeded".into(),
                request_id: None,
            })).await.ok();
            return;
        }
    }
    run.await;
}