我哭料!!!

Cargo.toml

@@ -26,4 +26,7 @@ serde_derive = "1.0"
 serde = { version = "1.0", features = ["derive"] } # 序列化
 rayon = "1.1" # 多线程
 ndarray = "0.13.0" # ndarray
-ws = "0.7.8"
+wsq = '0.9.1'
+crossbeam = "0.7"
+crossbeam-channel = "0.4"
+crossbeam-utils = "0.7"

src/main.rs

@@ -3,23 +3,30 @@ pub mod qawebsocket;
 pub mod qaeventmq;
 
 extern crate ndarray;
-use ws::{connect};
+use wsq::{connect, WebSocket, Error, ErrorKind};
+
 extern crate chrono;
 use chrono::prelude::*;
 use ndarray::array;
-use std::sync::mpsc::channel;
-use std::thread;
 
+use std::thread;
+use std::borrow::BorrowMut;
+extern crate crossbeam_utils;
+#[macro_use]
+extern crate crossbeam_channel;
+use crossbeam_channel::bounded;
+use crossbeam_utils::thread::scope;
 
 
 fn main() {
 
 
- let (ws_event_tx, ws_event_rx) = channel();
+ let (s1, r1) = bounded(0);
+ let (s2, r2) = (s1.clone(), r1.clone());
  let user_name = "000001".to_string();
 
  {
- let event_tx = ws_event_tx.clone();
+
  thread::spawn(move || {
  let mut client = qaeventmq::QAEventMQ{
  amqp: "amqp:https://admin:[email protected]:5672/".to_string(),
@@ -28,22 +35,68 @@ fn main() {
  routing_key: user_name.clone(),
  };
  println!("xxx");
- qaeventmq::QAEventMQ::consume(client, event_tx).unwrap();
+ qaeventmq::QAEventMQ::consume(client, s1).unwrap();
  });
  }
  let user_name = "000001".to_string();
- thread::spawn(|| {
- connect("ws:https://www.yutiansut.com:7788", move |out| {
- qawebsocket::QAtradeR{
- user_name: user_name.clone(),
- password: user_name.clone(),
- broker:"QUANTAXIS".to_string(),
- out:out
- }
-
- }).unwrap()
+
+
+
+//
+// connect("ws:https://101.132.37.31:7988", |out| {
+// qawebsocket::QAtradeR{
+// user_name: user_name.clone(),
+// password: user_name.clone(),
+// broker:"QUANTAXIS".to_string(),
+// out:out,
+// recv: r1.clone()
+// }});
+////
+ let mut ws = WebSocket::new(move |out| {
+ qawebsocket::QAtradeR{
+ user_name: user_name.clone(),
+ password: user_name.clone(),
+ broker:"QUANTAXIS".to_string(),
+ out:out,
+
+ }}
+
+ ).unwrap();
+
+
+ let sender = ws.handler.sender().clone();
+ thread::spawn(move||{
+
+ loop{
+ let data = r1.recv().unwrap();
+
+ println!("receive !!{:?}",data);
+ sender.send(format!("{}", data));
+
+ }
+
  });
- qawebsocket::QAtradeR::start(ws_event_rx);
+ let parsed ="ws:https://101.132.37.31:7988";
+ ws.connect(parsed.parse().unwrap()).unwrap();
+
+
+ ws.handler.run(ws.poll.borrow_mut());
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
  println!("trader");
 
 

src/qaeventmq.rs

@@ -4,8 +4,8 @@ use amiquip::{
  QueueDeclareOptions, Result, Publish
 };
 use amiquip::Delivery;
-use std::sync::mpsc::Sender;
 
+use crossbeam_channel::Sender;
 pub struct QAEventMQ{pub amqp : String,
  pub exchange: String,
  pub model: String,

src/qawebsocket.rs

@@ -2,11 +2,10 @@
 use serde::{Serialize, Deserialize};
 use std::time::Duration;
 use std::thread;
-use std::sync::mpsc::Receiver;
 
-use ws::{connect, Handler, Handshake, Message, Result};
-use ws::Sender;
 
+use wsq::{connect, Handler, Handshake, Message, Result};
+use wsq::Sender;
 
 pub struct QAtradeR{
  pub out: Sender,
@@ -16,15 +15,7 @@ pub struct QAtradeR{
 
 }
 
-impl QAtradeR{
- pub fn start(&mut self,event_rx: Receiver<String>){
- loop {
- let data = event_rx.recv().unwrap();
- println!("{:?}",data);
- self.out.send(format!("{}", data));
- }
- }
-}
+
 
 impl Handler for QAtradeR{
 
@@ -39,6 +30,7 @@ impl Handler for QAtradeR{
  password: self.password.clone(),};
 
  let b = serde_json::to_string(&login).unwrap();
+ println!("{}", b);
  self.out.send(b);
 
  Ok(())
@@ -49,6 +41,7 @@ impl Handler for QAtradeR{
 
  fn on_message(&mut self, msg: Message) -> Result<()> {
  if let Message::Text(message_text) = msg {
+ println!("{}", message_text);
  let peek = Peek { aid: "peek_message".to_string()};
  let b = serde_json::to_string(&peek).unwrap();
  self.out.send(b);