apca 0.26.1

A crate for interacting with the Alpaca API.
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
121
122

// Copyright (C) 2019-2023 The apca Developers
// SPDX-License-Identifier: GPL-3.0-or-later

use url::Url;

use tokio::net::TcpStream;

use tracing::debug;
use tracing::span;
use tracing::trace;
use tracing::Level;
use tracing_futures::Instrument;

use tungstenite::connect_async;
use tungstenite::MaybeTlsStream;
use tungstenite::WebSocketStream;

use websocket_util::wrap::Wrapper;

use crate::Error;


/// A custom [`Result`]-style type that we can implement a foreign trait
/// on.
#[derive(Debug)]
#[doc(hidden)]
pub enum MessageResult<T, E> {
  /// The success value.
  Ok(T),
  /// The error value.
  Err(E),
}

impl<T, E> From<Result<T, E>> for MessageResult<T, E> {
  #[inline]
  fn from(result: Result<T, E>) -> Self {
    match result {
      Ok(t) => Self::Ok(t),
      Err(e) => Self::Err(e),
    }
  }
}


/// Internal function to connect to websocket server.
async fn connect_internal(url: &Url) -> Result<WebSocketStream<MaybeTlsStream<TcpStream>>, Error> {
  let span = span!(Level::DEBUG, "stream");

  async move {
    debug!(message = "connecting", url = display(url));

    // We just ignore the response & headers that are sent along after
    // the connection is made. Alpaca does not seem to be using them,
    // really.
    let (stream, response) = connect_async(url).await?;
    debug!("connection successful");
    trace!(response = debug(&response));

    Ok(stream)
  }
  .instrument(span)
  .await
}


/// Connect to a websocket server.
pub(crate) async fn connect(
  url: &Url,
) -> Result<Wrapper<WebSocketStream<MaybeTlsStream<TcpStream>>>, Error> {
  connect_internal(url)
    .await
    .map(|stream| Wrapper::builder().build(stream))
}


#[cfg(test)]
pub(crate) mod test {
  use super::*;

  use std::future::Future;

  use websocket_util::test::mock_server;
  use websocket_util::test::WebSocketStream;
  use websocket_util::tungstenite::Error as WebSocketError;

  use crate::subscribable::Subscribable;
  use crate::ApiInfo;


  /// The fake key-id we use.
  pub(crate) const KEY_ID: &str = "USER12345678";
  /// The fake secret we use.
  pub(crate) const SECRET: &str = "justletmein";


  /// Instantiate a dummy websocket server serving messages as per the
  /// provided function `f` and attempt to connect to it to stream
  /// messages.
  pub(crate) async fn mock_stream<S, F, R>(f: F) -> Result<(S::Stream, S::Subscription), Error>
  where
    S: Subscribable<Input = ApiInfo>,
    F: FnOnce(WebSocketStream) -> R + Send + Sync + 'static,
    R: Future<Output = Result<(), WebSocketError>> + Send + Sync + 'static,
  {
    let addr = mock_server(f).await;
    let stream_url = Url::parse(&format!("ws://{addr}")).unwrap();

    // We just set both the API stream URL and the data stream URL to
    // our websocket server. We don't know which one clients are trying
    // to mock, but currently it's only one or the other.
    let api_info = ApiInfo {
      api_base_url: Url::parse("http://example.com").unwrap(),
      api_stream_url: stream_url.clone(),
      data_base_url: Url::parse("http://example.com").unwrap(),
      data_stream_base_url: stream_url.clone(),
      key_id: KEY_ID.to_string(),
      secret: SECRET.to_string(),
    };

    S::connect(&api_info).await
  }
}