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


use super::message::MessageData;
use super::error::StreamError;

use std::marker::PhantomData;
use tokio::sync::mpsc;

use serde::{Serialize, de::DeserializeOwned};

pub struct RawStreamer {
	inner: InnerRawStreamer
}

enum InnerRawStreamer {
	Sender(mpsc::Sender<MessageData>),
	Receiver(mpsc::Receiver<MessageData>)
}

impl RawStreamer {
	pub(crate) fn sender(tx: mpsc::Sender<MessageData>) -> Self {
		Self {
			inner: InnerRawStreamer::Sender(tx)
		}
	}

	pub(crate) fn receiver(rx: mpsc::Receiver<MessageData>) -> Self {
		Self {
			inner: InnerRawStreamer::Receiver(rx)
		}
	}

	pub fn assign_message<M>(self) -> Streamer<M> {
		Streamer {
			inner: self.inner,
			message: PhantomData
		}
	}
}

pub struct Streamer<M> {
	inner: InnerRawStreamer,
	message: PhantomData<M>
}

impl<M> Streamer<M> {
	/// ## Panics
	/// If you call this when the Stream::KIND is not Receiver
	pub async fn send(&mut self, data: M) -> Result<(), StreamError>
	where M: Serialize {
		match &mut self.inner {
			InnerRawStreamer::Sender(tx) => {
				tx.send(
					MessageData::serialize(data)
						.map_err(StreamError::Json)?
				).await
					.map_err(|_| StreamError::Closed)
			},
			_ => panic!("Streamer: cannot send, in receive mode")
		}
	}

	/// Completes when the receiver has dropped.
	///
	/// ## Panics
	/// If you call this when the Stream::KIND is not Receiver
	pub async fn closed(&self) {
		match &self.inner {
			InnerRawStreamer::Sender(tx) => {
				tx.closed().await;
			},
			_ => panic!("Streamer: cannot send, in receive mode")
		}
	}

	/// ## Panics
	/// If you call this when the Stream::KIND is not Sender
	pub async fn recv(&mut self) -> Result<M, StreamError>
	where M: DeserializeOwned {
		match &mut self.inner {
			InnerRawStreamer::Receiver(rx) => {
				let data = rx.recv().await
					.ok_or(StreamError::Closed)?;

				data.deserialize().map_err(StreamError::Json)
			},
			_ => panic!("Streamer: cannot receive, in sender mode")
		}
	}
}