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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
/*
use super::rpc::{Request, Response};
use super::storage::Storage;
use crate::session::{Metadata, Session};
use futures::channel::{mpsc, oneshot};
use futures::future::{FutureExt, Shared};
use std::collections::HashMap;
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use thiserror::Error;
use url::{ParseError, Url};
use ws::{Factory, Handler, Handshake, Message, Sender, WebSocket};
#[derive(Debug)]
pub struct Client {
sender: Sender,
event_loop: JoinHandle<Option<ws::Error>>,
connected: Shared<oneshot::Receiver<()>>,
context: Arc<Mutex<Context>>,
}
#[derive(Debug)]
struct Context {
session: Storage<Session>,
requests: HashMap<i32, oneshot::Sender<Response>>,
}
pub const DEFAULT_BRIDGE_URL: &str = "https://bridge.walletconnect.org";
impl Client {
pub fn new(profile: impl AsRef<Path>, meta: Metadata) -> Result<Self, CreationError> {
Client::with_bridge(profile, DEFAULT_BRIDGE_URL, meta)
}
pub fn with_bridge(
profile: impl AsRef<Path>,
bridge: impl AsRef<str>,
meta: Metadata,
) -> Result<Self, CreationError> {
let bridge = Url::parse(bridge.as_ref())?;
// NOTE: WalletConnect bridge URLs are expected to be automatically
// converted from a `http(s)` to `ws(s)` protocol for the WebSocket
// connection.
let rpc_url = {
let mut rpc_url = bridge.clone();
match rpc_url.scheme() {
"http" => rpc_url.set_scheme("ws").unwrap(),
"https" => rpc_url.set_scheme("wss").unwrap(),
"ws" | "wss" => {}
scheme => return Err(CreationError::BadScheme(scheme.into())),
}
rpc_url
};
let context = {
let session = Storage::for_session(profile.as_ref(), bridge, meta);
Arc::new(Mutex::new(Context {
session,
requests: HashMap::new(),
}))
};
let (connected_tx, connected_rx) = oneshot::channel();
let handler = ClientHandler {
context: context.clone(),
connected: Some(connected_tx),
};
let factory = ClientFactory(Some(handler));
let mut socket = WebSocket::new(factory)?;
socket.connect(rpc_url)?;
let sender = socket.broadcaster();
let event_loop = thread::spawn(move || socket.run().err());
Ok(Client {
sender,
event_loop,
context,
connected: connected_rx.shared(),
})
}
pub fn session(&self) -> Session {
self.context.lock().unwrap().session.clone()
}
pub fn close(self) -> Result<(), CloseError> {
self.sender.shutdown()?;
// NOTE: Intentionally propagate the event loop panic, as it is not
// intended to do so.
let err = self.event_loop.join().unwrap();
if let Some(err) = err {
Err(err.into())
} else {
Ok(())
}
}
}
struct ClientFactory(Option<ClientHandler>);
impl Factory for ClientFactory {
type Handler = ClientHandler;
fn connection_made(&mut self, _: Sender) -> Self::Handler {
self.0
.take()
.expect("more than one connection made for a single client")
}
}
struct ClientHandler {
context: Arc<Mutex<Context>>,
connected: Option<oneshot::Sender<()>>,
}
impl Handler for ClientHandler {
fn on_open(&mut self, _: Handshake) -> ws::Result<()> {
self.connected
.take()
.expect("connection opened more than once")
.send(())
.expect("client should not be dropped");
Ok(())
}
fn on_message(&mut self, msg: Message) -> ws::Result<()> {
let response = Response::parse(msg.as_text()?);
let mut context = self.context.lock().unwrap();
if let Some(sender) = context.requests.remove(&response.id) {
// NOTE: ignore errors where the receiver is dropped.
let _ = sender.send(response);
}
Ok(())
}
}
#[derive(Debug, Error)]
pub enum CreationError {
#[error("failed to parse URL: {0}")]
Parse(#[from] ParseError),
#[error("invalid URL scheme '{0}', must be 'http(s)' or 'ws(s)'")]
BadScheme(String),
#[error("IO error when creating the WebSocket: {0}")]
Io(#[from] ws::Error),
}
#[derive(Debug, Error)]
pub enum CloseError {
#[error("failed to shutdown WebSocket: {0}")]
Shutdown(#[from] ws::Error),
}
*/