eosio_shipper_gf/
lib.rs

1use error_chain::ChainedError;
2use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
3use futures_util::{future, pin_mut, SinkExt, StreamExt};
4use rs_abieos_gf::{abieos, Abieos};
5//use log::*;
6//use std::io::prelude::*;
7use rust_embed::RustEmbed;
8use tokio_tungstenite::connect_async;
9use tokio_tungstenite::tungstenite::{Bytes, Message};
10use url::Url;
11// `error_chain!` can recurse deeply
12//#![recursion_limit = "1024"]
13//
14use errors::{Error, ErrorKind, Result};
15#[macro_use]
16extern crate lazy_static;
17pub mod errors;
18pub mod shipper_types;
19
20use crate::shipper_types::{ShipRequests, ShipResultsEx};
21
22#[derive(RustEmbed)]
23#[folder = "resources/"]
24pub struct ShipAbiFiles;
25
26pub const EOSIO_SYSTEM: &str = "eosio";
27
28pub async fn get_sink_stream(
29    server_url: &str,
30    mut in_tx: UnboundedReceiver<ShipRequests>,
31    out_rx: UnboundedSender<ShipResultsEx>,
32) -> Result<()> {
33    let r = connect_async(server_url).await?;
34    let socket = r.0;
35    let (mut sink, mut stream) = socket.split();
36    match stream.next().await {
37        Some(msg) => {
38            let msg_text = msg
39                .map_err(|e| {
40                    Error::with_chain(e, "get_sink_stream fail")
41                })?
42                .into_text()
43                .map_err(|e| {
44                    Error::with_chain(e, "get_sink_stream into_text")
45                })?;
46            let shipper_abi = Abieos::new();
47            shipper_abi.set_abi_json(EOSIO_SYSTEM, msg_text.to_string()).map_err(|e| {
48                
49                Error::new(ErrorKind::Msg("Error parse ABI".parse().unwrap()), error_chain::State::default())
50            })?;
51
52            let out_loop = async {
53                loop {
54                    let data = stream
55                        .next()
56                        .await
57                        .unwrap()
58                        .expect("get_status_request_v0 Response error")
59                        .into_data();
60
61                    let r = ShipResultsEx::from_bin(&shipper_abi, &data).unwrap();
62
63                    out_rx.unbounded_send(r).expect("Didn't send");
64                }
65            };
66            let in_loop = async {
67                loop {
68                    let data: ShipRequests = in_tx.next().await.unwrap();
69
70                    match data {
71                        ShipRequests::get_status_request_v0(r) => {
72                            let req = r.to_bin(&shipper_abi).unwrap();
73                            let msg = Message::Binary(Bytes::from(req));
74                            sink.send(msg).await.expect("Didn't send");
75                        }
76                        ShipRequests::get_blocks_request_v0(br) => {
77                            let req = br.to_bin(&shipper_abi).unwrap();
78                            let msg = Message::Binary(Bytes::from(req));
79                            sink.send(msg).await.expect("Didn't send");
80                        }
81                        ShipRequests::get_blocks_ack_request_v0(ar) => {
82                            let req = ar.to_bin(&shipper_abi).unwrap();
83                            let msg = Message::Binary(Bytes::from(req));
84                            sink.send(msg).await.expect("Didn't send");
85                        }
86                        ShipRequests::quit => {
87                            eprintln!("QUIT");
88                            &sink.close();
89                            break;
90                        }
91                    }
92                }
93            };
94            pin_mut!(in_loop, out_loop);
95            future::join(in_loop, out_loop).await;
96
97            Ok(())
98        }
99        None => {
100            Err(ErrorKind::ExpectedABI.into())
101        }
102    }
103}
eosio_shipper_gf/lib.rs

eosio_shipper_gf/
lib.rs
