commit pre-zmq

Author: don

Cargo.lock

@@ -36,7 +36,6 @@ warp = "0.2"
 reqwest = "0.10.8"
 resiter = "0.4.0"
 im = "15.0.0"
-libzmq = "0.2.5"
 
 [dev-dependencies]
-pretty_assertions = "0.6.1"
+pretty_assertions = "0.6.1"

Cargo.toml

@@ -149,8 +149,9 @@ async fn get_logs(
                         timestamp: event.timestamp,
                         token: None,
                     });
-                    if let Err(err) = result {
-                        error!("Some error sending event over channel {}", err)
+                    match result {
+                        Ok(_) => {}
+                        Err(err) => error!("Some error sending event over channel {}", err),
                     }
                 });
             }

src/aws/cloudwatch_logs/mod.rs

@@ -5,80 +5,29 @@ extern crate log;
 
 
 use std::collections::HashMap;
-use std::str::Utf8Error;
 use std::sync::Arc;
 use std::sync::Mutex;
 
-use anyhow::Context as AnyhowContext;
-use futures::SinkExt;
-use tokio::stream::StreamExt;
-use tokio::time::{delay_for, Duration};
-use warp::Filter;
+use warp::{Filter};
 use warp::hyper::Method;
-use libzmq::{prelude::*, *};
 
 use aws::ecs::get_ecs_filter;
-use error::handle_rejection;
 
 use crate::aws::cloudwatch_logs::{get_logs_events_filter, get_logs_filter};
 use crate::aws::cloudwatch_logs::dto::LogsOptions;
 use crate::aws::dto::AwsRequest;
 use crate::aws::manager::setup_default_manager;
-use crate::notifications::{build_fan_notifications, NotUtf8, subscriber_connected};
-use libzmq::InprocAddr;
-use std::thread;
+use error::handle_rejection;
+use crate::notifications::{subscriber_connected, build_fan_notifications, NotUtf8};
 
-mod test;
 mod aws;
 mod error;
 mod notifications;
 
 #[tokio::main]
-async fn main() -> Result<(), anyhow::Error> {
+async fn main() {
     pretty_env_logger::init();
 
-    let inproc_addr: InprocAddr = InprocAddr::new_unique();
-    let inproc_socket = ServerBuilder::new().bind(&inproc_addr).build()?;
-    let subscriber_addr: TcpAddr = "eth0;192.168.1.1:5555".try_into()?;
-    let publish_socket = ClientBuilder::new().connect(subscriber_addr).build()?;
-
-    // Spawn the server thread.
-    let handle = thread::spawn(move || -> Result<(), Error> {
-        loop {
-            let request = inproc_socket.recv_msg()?;
-            let msg = request.to_str().unwrap_or_default();
-            info!("Received a message on the inproc socket, msg: {}", msg);
-            publish_socket.send(msg).unwrap_or_default();
-        }
-    });
-
-    let client = ClientBuilder::new().connect(inproc_addr).build()?;
-
-
-
-    // let publish_notification = context.socket(zmq::PUB)
-    //     .with_context(|| "Failed to instantiate notification socket!").unwrap();
-    // publish_notification
-    //     .bind("tcp://127.0.0.1:3031").unwrap();
-    //
-    // let inproc_socket = context.socket(zmq::SUB)
-    //     .with_context(|| "Failed to instantiate inproc socket!").unwrap();
-    // inproc_socket
-    //     .connect("inproc://internal_proxy").unwrap();
-    // inproc_socket
-    //     .set_subscribe(b"")
-    //     .expect("failed setting subscription");
-    //
-    // std::thread::spawn(move || {
-    //     while let Ok(msg) = inproc_socket.recv_msg(0) {
-    //         info!("Received a message on the inproc socket");
-    //         let result = publish_notification.send(msg, 0);
-    //         if let Err(err) = result {
-    //             error!("Error sending message to notification sender!, err {}", err)
-    //         }
-    //     }
-    // });
-
 
     let subscribers = Arc::new(Mutex::new(HashMap::new()));
     let subscribers = warp::any().map(move || subscribers.clone());
@@ -138,12 +87,6 @@ async fn main() -> Result<(), anyhow::Error> {
             warp::reply()
         });
 
-    let socket_test = warp::path("sockettest")
-        .and(warp::post())
-        .and(warp::body::content_length_limit(500))
-        .and(warp::body::bytes())
-        .and_then(move r|msg| notifications::handle_message(msg, client.clone()));
-
     let notifications = warp::path("notifications")
         .and(warp::get())
         .and(subscribers)
@@ -158,19 +101,9 @@ async fn main() -> Result<(), anyhow::Error> {
             .or(log_stream)
             .or(bootstrap_config)
             .or(notify)
-            .or(socket_test)
             .or(notifications)
             .with(cors)
             .recover(handle_rejection)
     )
         .run(([127, 0, 0, 1], 3030)).await;
-
-    // This will cause the server to fail with `InvalidCtx`.
-    Ctx::global().shutdown();
-
-    // Join with the thread.
-    let err = handle.join().unwrap().unwrap_err();
-    assert_eq!(err.kind(), ErrorKind::InvalidCtx);
-
-    Ok(())
 }

src/main.rs

@@ -3,12 +3,10 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Mutex;
-use libzmq::{prelude::*, *};
 
-use futures::{Stream, StreamExt, SinkExt};
+use futures::{Stream, StreamExt};
 use tokio::sync::mpsc;
-use warp::{sse::ServerSentEvent, Rejection};
-use anyhow::Context as AnyhowContext;
+use warp::{sse::ServerSentEvent};
 
 // Our global unique user id counter.
 pub static NEXT_USER_ID: AtomicUsize = AtomicUsize::new(1);
@@ -94,16 +92,3 @@ pub fn build_fan_notifications(msg: String, users: &Subscribers) {
             }
         });
 }
-
-pub async fn handle_message(bytes: bytes::Bytes, mut client: Client) -> Result<impl warp::Reply, Rejection> {
-    let msg =  std::str::from_utf8(&bytes);
-    info!("sending");
-    if let Ok(msg) = msg {
-        info!("msg was ok");
-        if let Err(err) = client.send(msg) {
-            error!("failed to send to internal proxy!, {}", err)
-        }
-    }
-
-    Ok(warp::reply())
-}