2021-03-05 02:37:51 +00:00
|
|
|
use crate::asb::Rate;
|
2021-03-08 23:39:33 +00:00
|
|
|
use anyhow::{Context, Result};
|
2021-03-05 02:37:51 +00:00
|
|
|
use bitcoin::util::amount::ParseAmountError;
|
|
|
|
use futures::{SinkExt, StreamExt};
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use serde_json::Value;
|
|
|
|
use std::convert::TryFrom;
|
|
|
|
use tokio::sync::watch;
|
2021-03-05 02:40:39 +00:00
|
|
|
use tokio_tungstenite::tungstenite;
|
2021-03-05 02:37:51 +00:00
|
|
|
|
2021-03-09 00:03:24 +00:00
|
|
|
type RateUpdate = Result<Rate, Error>;
|
|
|
|
|
2021-03-05 02:52:24 +00:00
|
|
|
pub async fn connect() -> Result<RateUpdateStream> {
|
2021-03-05 02:37:51 +00:00
|
|
|
let (rate_update, rate_update_receiver) = watch::channel(Err(Error::NotYetRetrieved));
|
|
|
|
|
2021-03-08 23:39:33 +00:00
|
|
|
let (rate_stream, _) = tokio_tungstenite::connect_async("wss://ws.kraken.com")
|
|
|
|
.await
|
|
|
|
.context("Failed to connect to Kraken websocket API")?;
|
2021-03-05 02:37:51 +00:00
|
|
|
|
|
|
|
let (mut rate_stream_sink, mut rate_stream) = rate_stream.split();
|
|
|
|
|
|
|
|
tokio::spawn(async move {
|
|
|
|
while let Some(msg) = rate_stream.next().await {
|
|
|
|
let msg = match msg {
|
2021-03-05 02:40:39 +00:00
|
|
|
Ok(tungstenite::Message::Text(msg)) => msg,
|
|
|
|
Ok(tungstenite::Message::Close(close_frame)) => {
|
|
|
|
if let Some(tungstenite::protocol::CloseFrame { code, reason }) = close_frame {
|
2021-03-08 23:23:12 +00:00
|
|
|
tracing::error!(
|
2021-03-05 02:37:51 +00:00
|
|
|
"Kraken rate stream was closed with code {} and reason: {}",
|
2021-03-08 23:23:12 +00:00
|
|
|
code,
|
|
|
|
reason
|
2021-03-05 02:37:51 +00:00
|
|
|
);
|
|
|
|
} else {
|
2021-03-08 23:23:12 +00:00
|
|
|
tracing::error!("Kraken rate stream was closed without code and reason");
|
2021-03-05 02:37:51 +00:00
|
|
|
}
|
2021-03-05 02:40:39 +00:00
|
|
|
let _ = rate_update.send(Err(Error::ConnectionClosed));
|
2021-03-05 02:37:51 +00:00
|
|
|
continue;
|
|
|
|
}
|
|
|
|
Ok(msg) => {
|
2021-03-08 23:23:12 +00:00
|
|
|
tracing::trace!(
|
2021-03-05 02:37:51 +00:00
|
|
|
"Kraken rate stream returned non text message that will be ignored: {}",
|
|
|
|
msg
|
|
|
|
);
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
Err(e) => {
|
2021-03-08 23:23:12 +00:00
|
|
|
tracing::error!(%e, "Error when reading from Kraken rate stream");
|
2021-03-05 02:37:51 +00:00
|
|
|
let _ = rate_update.send(Err(e.into()));
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
2021-03-05 02:40:39 +00:00
|
|
|
let update = match serde_json::from_str::<Event>(&msg) {
|
|
|
|
Ok(Event::SystemStatus) => {
|
|
|
|
tracing::debug!("Connected to Kraken websocket API");
|
2021-03-05 02:37:51 +00:00
|
|
|
continue;
|
|
|
|
}
|
2021-03-05 02:40:39 +00:00
|
|
|
Ok(Event::SubscriptionStatus) => {
|
|
|
|
tracing::debug!("Subscribed to updates for ticker");
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
Ok(Event::Heartbeat) => {
|
|
|
|
tracing::trace!("Received heartbeat message");
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
// if the message is not an event, it is a ticker update or an unknown event
|
|
|
|
Err(_) => match serde_json::from_str::<TickerUpdate>(&msg) {
|
|
|
|
Ok(ticker) => ticker,
|
|
|
|
Err(e) => {
|
|
|
|
tracing::warn!(%e, "Failed to deserialize message '{}' as ticker update", msg);
|
|
|
|
let _ = rate_update.send(Err(Error::UnknownMessage { msg }));
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
},
|
2021-03-05 02:37:51 +00:00
|
|
|
};
|
|
|
|
|
2021-03-05 02:40:39 +00:00
|
|
|
let rate = match Rate::try_from(update) {
|
2021-03-05 02:37:51 +00:00
|
|
|
Ok(rate) => rate,
|
|
|
|
Err(e) => {
|
|
|
|
let _ = rate_update.send(Err(e));
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
};
|
|
|
|
|
|
|
|
let _ = rate_update.send(Ok(rate));
|
|
|
|
}
|
|
|
|
});
|
|
|
|
|
|
|
|
rate_stream_sink
|
|
|
|
.send(SUBSCRIBE_XMR_BTC_TICKER_PAYLOAD.into())
|
|
|
|
.await?;
|
|
|
|
|
2021-03-05 02:52:24 +00:00
|
|
|
Ok(RateUpdateStream {
|
|
|
|
inner: rate_update_receiver,
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Clone, Debug)]
|
|
|
|
pub struct RateUpdateStream {
|
2021-03-09 00:03:24 +00:00
|
|
|
inner: watch::Receiver<RateUpdate>,
|
2021-03-05 02:52:24 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl RateUpdateStream {
|
2021-03-09 00:03:24 +00:00
|
|
|
pub async fn wait_for_update(&mut self) -> Result<RateUpdate> {
|
2021-03-05 02:52:24 +00:00
|
|
|
self.inner.changed().await?;
|
|
|
|
|
|
|
|
Ok(self.inner.borrow().clone())
|
|
|
|
}
|
|
|
|
|
2021-03-09 00:03:24 +00:00
|
|
|
pub fn latest_update(&mut self) -> RateUpdate {
|
2021-03-05 02:52:24 +00:00
|
|
|
self.inner.borrow().clone()
|
|
|
|
}
|
2021-03-05 02:37:51 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
const SUBSCRIBE_XMR_BTC_TICKER_PAYLOAD: &str = r#"
|
|
|
|
{ "event": "subscribe",
|
|
|
|
"pair": [ "XMR/XBT" ],
|
|
|
|
"subscription": {
|
|
|
|
"name": "ticker"
|
|
|
|
}
|
|
|
|
}"#;
|
|
|
|
|
|
|
|
#[derive(Clone, Debug, thiserror::Error)]
|
|
|
|
pub enum Error {
|
|
|
|
#[error("Rate has not yet been retrieved from Kraken websocket API")]
|
|
|
|
NotYetRetrieved,
|
2021-03-05 02:40:39 +00:00
|
|
|
#[error("The Kraken server closed the websocket connection")]
|
|
|
|
ConnectionClosed,
|
|
|
|
#[error("Websocket: {0}")]
|
2021-03-05 02:37:51 +00:00
|
|
|
WebSocket(String),
|
2021-03-05 02:40:39 +00:00
|
|
|
#[error("Received unknown message from Kraken: {msg}")]
|
|
|
|
UnknownMessage { msg: String },
|
2021-03-05 02:37:51 +00:00
|
|
|
#[error("Data field is missing")]
|
|
|
|
DataFieldMissing,
|
|
|
|
#[error("Ask Rate Element is of unexpected type")]
|
|
|
|
UnexpectedAskRateElementType,
|
|
|
|
#[error("Ask Rate Element is missing")]
|
|
|
|
MissingAskRateElementType,
|
|
|
|
#[error("Bitcoin amount parse error: ")]
|
|
|
|
BitcoinParseAmount(#[from] ParseAmountError),
|
|
|
|
}
|
|
|
|
|
2021-03-05 02:40:39 +00:00
|
|
|
impl From<tungstenite::Error> for Error {
|
|
|
|
fn from(err: tungstenite::Error) -> Self {
|
2021-03-05 02:37:51 +00:00
|
|
|
Error::WebSocket(format!("{:#}", err))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-03-05 02:40:39 +00:00
|
|
|
#[derive(Debug, Serialize, Deserialize, PartialEq)]
|
|
|
|
#[serde(tag = "event")]
|
|
|
|
enum Event {
|
|
|
|
#[serde(rename = "systemStatus")]
|
|
|
|
SystemStatus,
|
|
|
|
#[serde(rename = "heartbeat")]
|
|
|
|
Heartbeat,
|
|
|
|
#[serde(rename = "subscriptionStatus")]
|
|
|
|
SubscriptionStatus,
|
2021-03-05 02:37:51 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
|
|
#[serde(transparent)]
|
|
|
|
struct TickerUpdate(Vec<TickerField>);
|
|
|
|
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
|
|
#[serde(untagged)]
|
|
|
|
enum TickerField {
|
|
|
|
Data(TickerData),
|
|
|
|
Metadata(Value),
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
|
|
struct TickerData {
|
|
|
|
#[serde(rename = "a")]
|
|
|
|
ask: Vec<RateElement>,
|
|
|
|
#[serde(rename = "b")]
|
|
|
|
bid: Vec<RateElement>,
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
|
|
#[serde(untagged)]
|
|
|
|
enum RateElement {
|
|
|
|
Text(String),
|
|
|
|
Number(u64),
|
|
|
|
}
|
|
|
|
|
|
|
|
impl TryFrom<TickerUpdate> for Rate {
|
|
|
|
type Error = Error;
|
|
|
|
|
|
|
|
fn try_from(value: TickerUpdate) -> Result<Self, Error> {
|
|
|
|
let data = value
|
|
|
|
.0
|
|
|
|
.iter()
|
|
|
|
.find_map(|field| match field {
|
|
|
|
TickerField::Data(data) => Some(data),
|
|
|
|
TickerField::Metadata(_) => None,
|
|
|
|
})
|
|
|
|
.ok_or(Error::DataFieldMissing)?;
|
|
|
|
let ask = data.ask.first().ok_or(Error::MissingAskRateElementType)?;
|
|
|
|
let ask = match ask {
|
|
|
|
RateElement::Text(ask) => {
|
|
|
|
bitcoin::Amount::from_str_in(ask, ::bitcoin::Denomination::Bitcoin)?
|
|
|
|
}
|
|
|
|
_ => return Err(Error::UnexpectedAskRateElementType),
|
|
|
|
};
|
|
|
|
|
|
|
|
Ok(Self { ask })
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
|
|
mod tests {
|
|
|
|
use super::*;
|
|
|
|
|
2021-03-05 02:40:39 +00:00
|
|
|
#[test]
|
|
|
|
fn can_deserialize_system_status_event() {
|
|
|
|
let event = r#"{"connectionID":14859574189081089471,"event":"systemStatus","status":"online","version":"1.8.1"}"#;
|
|
|
|
|
|
|
|
let event = serde_json::from_str::<Event>(event).unwrap();
|
|
|
|
|
|
|
|
assert_eq!(event, Event::SystemStatus)
|
|
|
|
}
|
|
|
|
|
|
|
|
#[test]
|
|
|
|
fn can_deserialize_subscription_status_event() {
|
|
|
|
let event = r#"{"channelID":980,"channelName":"ticker","event":"subscriptionStatus","pair":"XMR/XBT","status":"subscribed","subscription":{"name":"ticker"}}"#;
|
|
|
|
|
|
|
|
let event = serde_json::from_str::<Event>(event).unwrap();
|
|
|
|
|
|
|
|
assert_eq!(event, Event::SubscriptionStatus)
|
|
|
|
}
|
|
|
|
|
|
|
|
#[test]
|
|
|
|
fn deserialize_ticker_update() {
|
|
|
|
let message = r#"[980,{"a":["0.00440700",7,"7.35318535"],"b":["0.00440200",7,"7.57416678"],"c":["0.00440700","0.22579000"],"v":["273.75489000","4049.91233351"],"p":["0.00446205","0.00441699"],"t":[123,1310],"l":["0.00439400","0.00429900"],"h":["0.00450000","0.00450000"],"o":["0.00449100","0.00433700"]},"ticker","XMR/XBT"]"#;
|
2021-03-05 02:37:51 +00:00
|
|
|
|
2021-03-05 02:40:39 +00:00
|
|
|
let _ = serde_json::from_str::<TickerUpdate>(message).unwrap();
|
2021-03-05 02:37:51 +00:00
|
|
|
}
|
|
|
|
}
|