ntfy 0.9.1

Library for sending notifications over ntfy
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

use std::pin::Pin;
use std::task::{Context, Poll};

use futures_util::stream::{FusedStream, Stream, StreamExt};
use tokio::net::TcpStream;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async};
use tungstenite::protocol::Message;
use url::Url;

use super::builder::SubscriberBuilder;
use super::request::get_request_builder;
use crate::auth::Auth;
use crate::error::Error;
use crate::payload::ReceivedPayload;

/// Async subscriber
#[derive(Debug, Clone)]
pub struct Async {
    auth: Option<Auth>,
}

impl Async {
    #[inline]
    pub(crate) fn new(builder: SubscriberBuilder) -> Result<Self, Error> {
        Ok(Self { auth: builder.auth })
    }

    pub(crate) async fn subscribe(&self, url: &Url, topic: &str) -> Result<MessageStream, Error> {
        let builder = get_request_builder(url, topic, &self.auth)?;

        // Create message iterator
        Ok(MessageStream {
            socket: connect_async(builder).await?.0,
        })
    }
}

pub struct MessageStream {
    socket: WebSocketStream<MaybeTlsStream<TcpStream>>,
}

impl Stream for MessageStream {
    type Item = Result<ReceivedPayload, Error>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if self.socket.is_terminated() {
            return Poll::Ready(None);
        }

        let text_message = loop {
            let message = match self.socket.poll_next_unpin(cx) {
                Poll::Pending => return Poll::Pending,
                Poll::Ready(Some(Ok(message))) => message,
                Poll::Ready(Some(Err(error))) => return Poll::Ready(Some(Err(Error::from(error)))),
                Poll::Ready(None) => return Poll::Ready(None),
            };

            match message {
                Message::Close(_) => return Poll::Ready(None),
                Message::Text(text_message) => break text_message,
                _ => {}
            }
        };

        match serde_json::from_str(text_message.as_str()) {
            Ok(received_message) => Poll::Ready(Some(Ok(received_message))),
            Err(error) => Poll::Ready(Some(Err(Error::from(error)))),
        }
    }
}