Compare commits
No commits in common. "2793508a9371c0d1aa02499121a00620346870ca" and "e6d76ba37c4aa2fc9827241f70c95c7d82494548" have entirely different histories.
2793508a93
...
e6d76ba37c
5 changed files with 186 additions and 186 deletions
|
@ -1,6 +1,7 @@
|
||||||
use std::{
|
use std::{
|
||||||
collections::{HashMap, HashSet},
|
collections::{HashMap, HashSet},
|
||||||
sync::{Arc, Mutex},
|
io::BufRead,
|
||||||
|
sync::{mpsc::channel, Arc, Mutex},
|
||||||
thread,
|
thread,
|
||||||
time::Duration,
|
time::Duration,
|
||||||
};
|
};
|
||||||
|
@ -9,18 +10,28 @@ use nebkor_maelstrom::{protocol::Payload, Body, Message, Node, Runner};
|
||||||
use rand::Rng;
|
use rand::Rng;
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
|
let out = std::io::stdout();
|
||||||
|
let std_in = Arc::new(std::io::stdin());
|
||||||
|
|
||||||
|
let (tx, rx) = channel();
|
||||||
|
let input = tx.clone();
|
||||||
|
|
||||||
|
let i = thread::spawn(move || {
|
||||||
|
let g = std_in.lock();
|
||||||
|
for line in g.lines().map_while(Result::ok) {
|
||||||
|
if let Ok(msg) = serde_json::from_str::<Message>(&line) {
|
||||||
|
input.send(msg).unwrap();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
let node = BCaster::default();
|
let node = BCaster::default();
|
||||||
let node = Arc::new(Mutex::new(node));
|
let node = Arc::new(Mutex::new(node));
|
||||||
|
|
||||||
let runner = Runner::new(node);
|
let runner = Runner::new(node, out);
|
||||||
let runner = &runner;
|
let runner = &runner;
|
||||||
|
|
||||||
runner.run(Some(Box::new(on_init)));
|
let g = thread::spawn(move || loop {
|
||||||
}
|
|
||||||
|
|
||||||
fn on_init(runner: &Runner) {
|
|
||||||
let tx = runner.get_backdoor();
|
|
||||||
thread::spawn(move || loop {
|
|
||||||
let millis = rand::thread_rng().gen_range(400..=800);
|
let millis = rand::thread_rng().gen_range(400..=800);
|
||||||
thread::sleep(Duration::from_millis(millis));
|
thread::sleep(Duration::from_millis(millis));
|
||||||
let body = Body::from_type("do_gossip");
|
let body = Body::from_type("do_gossip");
|
||||||
|
@ -30,6 +41,10 @@ fn on_init(runner: &Runner) {
|
||||||
};
|
};
|
||||||
tx.send(msg).unwrap();
|
tx.send(msg).unwrap();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
runner.run(rx, None);
|
||||||
|
let _ = i.join();
|
||||||
|
let _ = g.join();
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Default)]
|
#[derive(Clone, Default)]
|
||||||
|
|
|
@ -1,4 +1,8 @@
|
||||||
use std::sync::{Arc, Mutex};
|
use std::{
|
||||||
|
io::BufRead,
|
||||||
|
sync::{mpsc::channel, Arc, Mutex},
|
||||||
|
thread,
|
||||||
|
};
|
||||||
|
|
||||||
use nebkor_maelstrom::{Body, Message, Node, Runner};
|
use nebkor_maelstrom::{Body, Message, Node, Runner};
|
||||||
|
|
||||||
|
@ -15,11 +19,26 @@ impl Node for Echo {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
|
let out = std::io::stdout();
|
||||||
|
let std_in = Arc::new(std::io::stdin());
|
||||||
|
|
||||||
|
let (tx, rx) = channel();
|
||||||
|
|
||||||
|
let i = thread::spawn(move || {
|
||||||
|
let g = std_in.lock();
|
||||||
|
for line in g.lines().map_while(Result::ok) {
|
||||||
|
if let Ok(msg) = serde_json::from_str::<Message>(&line) {
|
||||||
|
tx.send(msg).unwrap();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
let node = Echo;
|
let node = Echo;
|
||||||
|
|
||||||
let node = Arc::new(Mutex::new(node));
|
let node = Arc::new(Mutex::new(node));
|
||||||
|
|
||||||
let runner = Runner::new(node);
|
let runner = Runner::new(node, out);
|
||||||
|
|
||||||
runner.run(None);
|
runner.run(rx, None);
|
||||||
|
i.join().unwrap();
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,17 +1,17 @@
|
||||||
use std::sync::{Arc, Mutex};
|
use std::{
|
||||||
|
rc::Rc,
|
||||||
|
sync::{Arc, Mutex},
|
||||||
|
};
|
||||||
|
|
||||||
use nebkor_maelstrom::{mk_payload, Body, Message, Node, Runner};
|
use nebkor_maelstrom::{mk_payload, mk_stdin, Body, Message, Node, Runner};
|
||||||
|
|
||||||
const KEY: &str = "COUNTER";
|
|
||||||
|
|
||||||
#[derive(Clone, Default)]
|
|
||||||
struct Counter;
|
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
let node = Counter;
|
let node = Counter;
|
||||||
let node = Arc::new(Mutex::new(node));
|
let node = Arc::new(Mutex::new(node));
|
||||||
|
|
||||||
let runner = Runner::new(node);
|
let runner = Rc::new(Runner::new(node));
|
||||||
|
|
||||||
|
let (i, _, rx) = mk_stdin();
|
||||||
|
|
||||||
let on_init = |rnr: &Runner| {
|
let on_init = |rnr: &Runner| {
|
||||||
let payload = mk_payload(&[
|
let payload = mk_payload(&[
|
||||||
|
@ -26,56 +26,62 @@ fn main() {
|
||||||
|
|
||||||
let on_init = Box::new(on_init);
|
let on_init = Box::new(on_init);
|
||||||
|
|
||||||
runner.run(Some(on_init));
|
runner.run(rx, Some(on_init));
|
||||||
|
i.join().unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const KEY: &str = "COUNTER";
|
||||||
|
|
||||||
|
#[derive(Clone, Default)]
|
||||||
|
struct Counter;
|
||||||
|
|
||||||
impl Node for Counter {
|
impl Node for Counter {
|
||||||
fn handle<'slf>(&'slf mut self, runner: &'slf Runner, req: Message) {
|
fn handle<'slf>(&'slf mut self, runner: &'slf Rc<Runner>, req: Message) {
|
||||||
let read_payload = mk_payload(&[("key", KEY.into())]);
|
|
||||||
let read_body = Body::from_type("read").with_payload(read_payload);
|
|
||||||
let typ = req.body.typ.as_str();
|
let typ = req.body.typ.as_str();
|
||||||
//let frm = req.src.clone();
|
let frm = req.src.clone();
|
||||||
//let msg_id = req.body.msg_id.to_owned();
|
let msg_id = req.body.msg_id.to_owned();
|
||||||
match typ {
|
match typ {
|
||||||
"add" => {
|
"add" => {
|
||||||
|
let read_runner = runner.clone();
|
||||||
|
let cas_runner = runner.clone();
|
||||||
|
let add_req = req.clone();
|
||||||
|
|
||||||
|
let cas_handler = move |_msg: Message| {
|
||||||
|
let req = add_req;
|
||||||
|
cas_runner.reply(&req, Body::from_type("add_ok"));
|
||||||
|
};
|
||||||
|
|
||||||
let delta = req.body.payload.get("delta").unwrap().as_i64().unwrap();
|
let delta = req.body.payload.get("delta").unwrap().as_i64().unwrap();
|
||||||
let cur = runner
|
let read_handler = move |msg: Message| {
|
||||||
.rpc("seq-kv", read_body)
|
let value = msg.body.payload.get("value").unwrap().as_i64().unwrap();
|
||||||
.recv()
|
let payload = mk_payload(&[
|
||||||
.unwrap()
|
("from", value.into()),
|
||||||
.body
|
("to", (value + delta).into()),
|
||||||
.payload
|
|
||||||
.get("value")
|
|
||||||
.cloned()
|
|
||||||
.unwrap()
|
|
||||||
.as_i64()
|
|
||||||
.unwrap();
|
|
||||||
let cas_payload = mk_payload(&[
|
|
||||||
("key", KEY.into()),
|
("key", KEY.into()),
|
||||||
("from", cur.into()),
|
|
||||||
("to", (cur + delta).into()),
|
|
||||||
]);
|
]);
|
||||||
// ERRORS BE HERE
|
let body = Body::from_type("cas").with_payload(payload);
|
||||||
runner
|
read_runner.rpc("seq-kv", body, Box::new(cas_handler));
|
||||||
.rpc("seq-kv", Body::from_type("cas").with_payload(cas_payload))
|
};
|
||||||
.recv()
|
// kick it off by calling "read" on seq-kv:
|
||||||
.unwrap();
|
|
||||||
runner.reply(&req, Body::from_type("add_ok"));
|
|
||||||
}
|
|
||||||
"read" => {
|
|
||||||
let payload = mk_payload(&[("key", KEY.into())]);
|
let payload = mk_payload(&[("key", KEY.into())]);
|
||||||
let body = Body::from_type("read").with_payload(payload);
|
let body = Body::from_type("read").with_payload(payload);
|
||||||
let val = runner
|
runner.rpc("seq-kv", body, Box::new(read_handler));
|
||||||
.rpc("seq-kv", body)
|
}
|
||||||
.recv()
|
"read" => {
|
||||||
.unwrap()
|
let rn = runner.clone();
|
||||||
.body
|
let h = move |msg: Message| {
|
||||||
.payload
|
let src = frm.clone();
|
||||||
.get("value")
|
let value = msg.body.payload.get("value").unwrap().as_i64().unwrap();
|
||||||
.cloned()
|
let irt = msg_id;
|
||||||
.unwrap();
|
let payload = mk_payload(&[("value", value.into())]);
|
||||||
let body = Body::from_type("read_ok").with_payload(mk_payload(&[("value", val)]));
|
let body = Body::from_type("read_ok")
|
||||||
runner.reply(&req, body);
|
.with_in_reply_to(irt)
|
||||||
|
.with_payload(payload);
|
||||||
|
rn.send(&src, body);
|
||||||
|
};
|
||||||
|
let payload = mk_payload(&[("key", KEY.into())]);
|
||||||
|
let body = Body::from_type("read").with_payload(payload);
|
||||||
|
runner.rpc("seq-kv", body, Box::new(h));
|
||||||
}
|
}
|
||||||
_ => {
|
_ => {
|
||||||
eprintln!("unknown type: {req:?}");
|
eprintln!("unknown type: {req:?}");
|
||||||
|
|
|
@ -1,14 +1,33 @@
|
||||||
use std::sync::{Arc, Mutex};
|
use std::{
|
||||||
|
io::BufRead,
|
||||||
|
sync::{mpsc::channel, Arc, Mutex},
|
||||||
|
thread,
|
||||||
|
};
|
||||||
|
|
||||||
use nebkor_maelstrom::{protocol::Payload, Body, Message, Node, Runner};
|
use nebkor_maelstrom::{protocol::Payload, Body, Message, Node, Runner};
|
||||||
|
|
||||||
fn main() {
|
fn main() {
|
||||||
|
let out = std::io::stdout();
|
||||||
|
let std_in = Arc::new(std::io::stdin());
|
||||||
|
|
||||||
|
let (tx, rx) = channel();
|
||||||
|
|
||||||
|
let i = thread::spawn(move || {
|
||||||
|
let g = std_in.lock();
|
||||||
|
for line in g.lines().map_while(Result::ok) {
|
||||||
|
if let Ok(msg) = serde_json::from_str::<Message>(&line) {
|
||||||
|
tx.send(msg).unwrap();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
let node = GenUid;
|
let node = GenUid;
|
||||||
let node = Arc::new(Mutex::new(node));
|
let node = Arc::new(Mutex::new(node));
|
||||||
|
|
||||||
let runner = Runner::new(node);
|
let runner = Runner::new(node, out);
|
||||||
|
|
||||||
runner.run(None);
|
runner.run(rx, None);
|
||||||
|
i.join().unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Default)]
|
#[derive(Clone, Default)]
|
||||||
|
|
|
@ -1,12 +1,13 @@
|
||||||
use std::{
|
use std::{
|
||||||
collections::HashMap,
|
collections::HashMap,
|
||||||
io::{BufRead, Write},
|
io::{BufRead, Stdout, Write},
|
||||||
|
rc::Rc,
|
||||||
sync::{
|
sync::{
|
||||||
atomic::{AtomicU64, AtomicUsize, Ordering},
|
atomic::{AtomicU64, AtomicUsize, Ordering},
|
||||||
mpsc::{channel, Receiver, Sender},
|
mpsc::{channel, Receiver, Sender},
|
||||||
Arc, Mutex, OnceLock,
|
Arc, LazyLock, Mutex, OnceLock,
|
||||||
},
|
},
|
||||||
thread::{self},
|
thread::{self, JoinHandle},
|
||||||
};
|
};
|
||||||
|
|
||||||
pub mod protocol;
|
pub mod protocol;
|
||||||
|
@ -17,57 +18,25 @@ use serde_json::Value;
|
||||||
pub mod kv;
|
pub mod kv;
|
||||||
|
|
||||||
pub type DynNode = Arc<Mutex<dyn Node>>;
|
pub type DynNode = Arc<Mutex<dyn Node>>;
|
||||||
|
pub type Handler = Box<dyn FnOnce(Message)>; // -> Result<Option<Value>>>;
|
||||||
pub type OnInit = Box<dyn Fn(&Runner)>;
|
pub type OnInit = Box<dyn Fn(&Runner)>;
|
||||||
|
|
||||||
pub type Result<T> = std::result::Result<T, ErrorCode>;
|
pub type Result<T> = std::result::Result<T, ErrorCode>;
|
||||||
|
|
||||||
pub type RpcPromise = Receiver<Message>;
|
|
||||||
|
|
||||||
static MSG_ID: AtomicU64 = AtomicU64::new(0);
|
static MSG_ID: AtomicU64 = AtomicU64::new(0);
|
||||||
|
static OUTPUT: LazyLock<Stdout> = LazyLock::new(std::io::stdout);
|
||||||
|
|
||||||
pub trait Node {
|
pub trait Node {
|
||||||
fn handle(&mut self, runner: &Runner, msg: Message);
|
fn handle(&mut self, runner: &Rc<Runner>, msg: Message);
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
|
||||||
pub struct Network {
|
|
||||||
promises: Arc<Mutex<HashMap<u64, Sender<Message>>>>,
|
|
||||||
node_output_tx: Sender<Message>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl Network {
|
|
||||||
pub fn new() -> (Self, Receiver<Message>) {
|
|
||||||
let (node_output_tx, node_output_rx) = channel();
|
|
||||||
let net = Self {
|
|
||||||
node_output_tx,
|
|
||||||
promises: Arc::new(Mutex::new(HashMap::default())),
|
|
||||||
};
|
|
||||||
(net, node_output_rx)
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn send(&self, msg: Message) {
|
|
||||||
self.node_output_tx.send(msg).unwrap();
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn rpc(&self, msg: Message) -> RpcPromise {
|
|
||||||
let (tx, rx) = channel();
|
|
||||||
{
|
|
||||||
let msg_id = msg.body.msg_id;
|
|
||||||
let mut g = self.promises.lock().unwrap();
|
|
||||||
g.insert(msg_id, tx);
|
|
||||||
}
|
|
||||||
self.send(msg);
|
|
||||||
rx
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Clone)]
|
||||||
pub struct Runner {
|
pub struct Runner {
|
||||||
node: DynNode,
|
node: DynNode,
|
||||||
node_id: OnceLock<String>,
|
node_id: OnceLock<String>,
|
||||||
nodes: OnceLock<Vec<String>>,
|
nodes: OnceLock<Vec<String>>,
|
||||||
network: OnceLock<Network>,
|
|
||||||
backdoor: OnceLock<Sender<Message>>,
|
|
||||||
steps: Arc<AtomicUsize>,
|
steps: Arc<AtomicUsize>,
|
||||||
|
handlers: Arc<Mutex<HashMap<u64, Handler>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Runner {
|
impl Runner {
|
||||||
|
@ -76,87 +45,44 @@ impl Runner {
|
||||||
node,
|
node,
|
||||||
nodes: OnceLock::new(),
|
nodes: OnceLock::new(),
|
||||||
node_id: OnceLock::new(),
|
node_id: OnceLock::new(),
|
||||||
network: OnceLock::new(),
|
|
||||||
backdoor: OnceLock::new(),
|
|
||||||
steps: Arc::new(AtomicUsize::new(0)),
|
steps: Arc::new(AtomicUsize::new(0)),
|
||||||
|
handlers: Default::default(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn run(&self, on_init: Option<OnInit>) {
|
pub fn run(self: &Rc<Self>, input: Receiver<Message>, on_init: Option<OnInit>) {
|
||||||
let (stdin_tx, stdin_rx) = channel();
|
for msg in input.iter() {
|
||||||
let (stdout_tx, stdout_rx) = channel();
|
let typ = &msg.body.typ;
|
||||||
|
if let "init" = typ.as_str() {
|
||||||
thread::spawn(move || {
|
|
||||||
let stdin = std::io::stdin().lock().lines();
|
|
||||||
for line in stdin {
|
|
||||||
stdin_tx.send(line.unwrap()).unwrap();
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
thread::spawn(move || {
|
|
||||||
let mut stdout = std::io::stdout().lock();
|
|
||||||
for msg in stdout_rx {
|
|
||||||
writeln!(&mut stdout, "{msg}").unwrap();
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
self.run_internal(stdout_tx, stdin_rx, on_init);
|
|
||||||
}
|
|
||||||
|
|
||||||
fn run_internal(
|
|
||||||
&self,
|
|
||||||
stdout_tx: Sender<String>,
|
|
||||||
stdin_rx: Receiver<String>,
|
|
||||||
on_init: Option<OnInit>,
|
|
||||||
) {
|
|
||||||
let (network, node_receiver) = Network::new();
|
|
||||||
let _ = self.network.get_or_init(|| network.clone());
|
|
||||||
self.run_output(stdout_tx, node_receiver);
|
|
||||||
self.run_input(stdin_rx, network, on_init);
|
|
||||||
}
|
|
||||||
|
|
||||||
fn run_input(&self, stdin_rx: Receiver<String>, network: Network, on_init: Option<OnInit>) {
|
|
||||||
let (json_tx, json_rx) = channel();
|
|
||||||
let _ = self.backdoor.get_or_init(|| json_tx.clone());
|
|
||||||
|
|
||||||
thread::spawn(move || {
|
|
||||||
for line in stdin_rx {
|
|
||||||
let msg: Message = serde_json::from_str(&line).unwrap();
|
|
||||||
let irt = msg.body.in_reply_to;
|
|
||||||
if let Some(tx) = network.promises.lock().unwrap().remove(&irt) {
|
|
||||||
tx.send(msg).unwrap();
|
|
||||||
} else {
|
|
||||||
json_tx.send(msg).unwrap();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
for msg in json_rx {
|
|
||||||
let mut node = self.node.lock().unwrap();
|
|
||||||
if msg.body.typ.as_str() == "init" {
|
|
||||||
self.init(&msg);
|
self.init(&msg);
|
||||||
let body = Body::from_type("init_ok");
|
let body = Body::from_type("init_ok");
|
||||||
self.reply(&msg, body);
|
self.reply(&msg, body);
|
||||||
if let Some(ref on_init) = on_init {
|
if let Some(ref h) = on_init {
|
||||||
on_init(self);
|
h(self);
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
node.handle(self, msg);
|
let irt = msg.body.in_reply_to;
|
||||||
|
{
|
||||||
|
let mut g = self.handlers.lock().unwrap();
|
||||||
|
if let Some(h) = g.remove(&irt) {
|
||||||
|
h(msg.clone());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let mut n = self.node.lock().unwrap();
|
||||||
|
n.handle(self, msg);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn run_output(&self, stdout_tx: Sender<String>, node_output_rx: Receiver<Message>) {
|
pub fn rpc(&self, dest: &str, body: Body, handler: Handler) {
|
||||||
thread::spawn(move || {
|
let mut body = body;
|
||||||
while let Ok(msg) = node_output_rx.recv() {
|
let msg_id = self.next_msg_id();
|
||||||
let msg = serde_json::to_string(&msg).unwrap();
|
body.msg_id = msg_id;
|
||||||
stdout_tx.send(msg).unwrap();
|
{
|
||||||
|
let mut g = self.handlers.lock().unwrap();
|
||||||
|
g.insert(msg_id, handler);
|
||||||
}
|
}
|
||||||
});
|
self.send(dest, body);
|
||||||
}
|
|
||||||
|
|
||||||
pub fn get_backdoor(&self) -> Sender<Message> {
|
|
||||||
self.backdoor.get().unwrap().clone()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn node_id(&self) -> String {
|
pub fn node_id(&self) -> String {
|
||||||
|
@ -198,10 +124,10 @@ impl Runner {
|
||||||
.unwrap()
|
.unwrap()
|
||||||
.iter()
|
.iter()
|
||||||
.map(|s| s.as_str().unwrap().to_string())
|
.map(|s| s.as_str().unwrap().to_string())
|
||||||
.collect();
|
.collect::<Vec<_>>();
|
||||||
|
|
||||||
let _ = self.node_id.get_or_init(|| node_id);
|
let _ = self.node_id.get_or_init(|| node_id.to_owned());
|
||||||
let _ = self.nodes.get_or_init(|| nodes);
|
let _ = self.nodes.get_or_init(|| nodes.to_vec());
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn reply(&self, req: &Message, body: Body) {
|
pub fn reply(&self, req: &Message, body: Body) {
|
||||||
|
@ -222,21 +148,36 @@ impl Runner {
|
||||||
dest: dest.to_string(),
|
dest: dest.to_string(),
|
||||||
body,
|
body,
|
||||||
};
|
};
|
||||||
self.network.get().unwrap().send(msg);
|
let msg = serde_json::to_string(&msg).unwrap();
|
||||||
|
self.writeln(&msg);
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn rpc(&self, dest: &str, body: Body) -> RpcPromise {
|
fn writeln(&self, msg: &str) {
|
||||||
let mut body = body;
|
let mut out = OUTPUT.lock();
|
||||||
if body.msg_id == 0 {
|
let msg = format!("{msg}\n");
|
||||||
body.msg_id = self.next_msg_id();
|
out.write_all(msg.as_bytes()).unwrap();
|
||||||
|
out.flush().unwrap();
|
||||||
}
|
}
|
||||||
let msg = Message {
|
|
||||||
src: self.node_id().to_string(),
|
|
||||||
dest: dest.to_string(),
|
|
||||||
body,
|
|
||||||
};
|
|
||||||
self.network.get().unwrap().rpc(msg)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Feeds lines from stdin to the MPSC Sender, so that the Receiver can be used
|
||||||
|
/// in the Runner::run() method. Clone the Sender if you want to inject messages
|
||||||
|
/// into the Runner. Join the handle after `run()`.
|
||||||
|
pub fn mk_stdin() -> (JoinHandle<()>, Sender<Message>, Receiver<Message>) {
|
||||||
|
let stdin = std::io::stdin();
|
||||||
|
let (tx, rx) = channel();
|
||||||
|
let xtra_input = tx.clone();
|
||||||
|
|
||||||
|
let join = thread::spawn(move || {
|
||||||
|
let g = stdin.lock();
|
||||||
|
for line in g.lines().map_while(std::result::Result::ok) {
|
||||||
|
if let Ok(msg) = serde_json::from_str(&line) {
|
||||||
|
tx.send(msg).unwrap();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
(join, xtra_input, rx)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn mk_payload(payload: &[(&str, Value)]) -> Payload {
|
pub fn mk_payload(payload: &[(&str, Value)]) -> Payload {
|
||||||
|
|
Loading…
Reference in a new issue