openwire 0.1.1

OkHttp-inspired async HTTP client for Rust built on hyper and tower
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

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

use futures_util::sink::Sink;
use futures_util::stream::Stream;

use openwire_core::websocket::{
    BoxEngineSink, BoxEngineStream, CloseInitiator, EngineFrame, MessageKind, WebSocketChannel,
    WebSocketEngineError,
};
use openwire_core::{CallContext, SharedEventListener};

pub(crate) fn instrument_channel(
    channel: WebSocketChannel,
    ctx: CallContext,
    listener: SharedEventListener,
) -> WebSocketChannel {
    WebSocketChannel {
        send: Box::pin(InstrumentedSink {
            inner: channel.send,
            ctx: ctx.clone(),
            listener: listener.clone(),
        }),
        recv: Box::pin(InstrumentedStream {
            inner: channel.recv,
            ctx,
            listener,
        }),
    }
}

struct InstrumentedSink {
    inner: BoxEngineSink,
    ctx: CallContext,
    listener: SharedEventListener,
}

impl Sink<EngineFrame> for InstrumentedSink {
    type Error = WebSocketEngineError;

    fn poll_ready(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.as_mut().poll_ready(cx)
    }

    fn start_send(mut self: Pin<&mut Self>, item: EngineFrame) -> Result<(), Self::Error> {
        match &item {
            EngineFrame::Text(text) => {
                self.listener
                    .websocket_message_sent(&self.ctx, MessageKind::Text, text.len())
            }
            EngineFrame::Binary(bytes) => {
                self.listener
                    .websocket_message_sent(&self.ctx, MessageKind::Binary, bytes.len())
            }
            EngineFrame::Ping(_) => self.listener.websocket_ping_sent(&self.ctx),
            EngineFrame::Pong(_) | EngineFrame::Close { .. } => {}
        }
        self.inner.as_mut().start_send(item)
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.as_mut().poll_flush(cx)
    }

    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.as_mut().poll_close(cx)
    }
}

struct InstrumentedStream {
    inner: BoxEngineStream,
    ctx: CallContext,
    listener: SharedEventListener,
}

impl Stream for InstrumentedStream {
    type Item = Result<EngineFrame, WebSocketEngineError>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        match self.inner.as_mut().poll_next(cx) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Ready(Some(Ok(frame))) => {
                match &frame {
                    EngineFrame::Text(text) => self.listener.websocket_message_received(
                        &self.ctx,
                        MessageKind::Text,
                        text.len(),
                    ),
                    EngineFrame::Binary(bytes) => self.listener.websocket_message_received(
                        &self.ctx,
                        MessageKind::Binary,
                        bytes.len(),
                    ),
                    EngineFrame::Pong(_) => self.listener.websocket_pong_received(&self.ctx),
                    EngineFrame::Close { code, reason } => self.listener.websocket_closing(
                        &self.ctx,
                        *code,
                        reason,
                        CloseInitiator::Remote,
                    ),
                    EngineFrame::Ping(_) => {}
                }
                Poll::Ready(Some(Ok(frame)))
            }
            Poll::Ready(Some(Err(error))) => Poll::Ready(Some(Err(error))),
        }
    }
}