Hyper everywhere (toshi-search#334)

* Hyper everywhere

* Hyper everywhere

* Some client clean up

* just gotta make the tests work

* just gotta make the tests work

* Most of the test work now, I just gotta do delete terms now

* Some style and convenience tweaks

* Make a test not fail

* Remote delete of terms from an index

Author: hntd187

Cargo.lock

@@ -19,28 +19,19 @@ members = [ "./", "toshi-proto" ]
 [dependencies]
 toshi-proto            = { path = "toshi-proto" }
 tower                  = "^0.1"
-tower-service          = "^0.2"
 tower-buffer           = "^0.1"
-tower-grpc             = { git = "https://github.com/tower-rs/tower-grpc" }
-tower-h2               = { git = "https://github.com/tower-rs/tower-h2" }
-tower-http             = { git = "https://github.com/tower-rs/tower-http" }
+tower-grpc             = { git = "https://github.com/tower-rs/tower-grpc", features = ["tower-hyper"] }
+tower-hyper            = { git = "https://github.com/tower-rs/tower-hyper" }
 tower-request-modifier = { git = "https://github.com/tower-rs/tower-http" }
 tower-consul           = { git = "https://github.com/LucioFranco/tower-consul" }
 futures-watch          = { git = "https://github.com/carllerche/better-future" }
-tower-web              = "^0.3"
 http                   = "^0.1"
-h2                     = "^0.1"
-flate2                 = "^1.0"
 bytes                  = "^0.4"
-prost                  = "^0.5"
-prost-derive           = "^0.5"
 hyper                  = "^0.12"
-hyper-tls              = "^0.3"
 serde_json             = "^1.0"
 futures                = "^0.1"
 tantivy                = "^0.9"
 tokio                  = "^0.1"
-tokio-executor         = "^0.1"
 tokio-signal           = "^0.2"
 config                 = "^0.9"
 log                    = "^0.4"

Cargo.toml

@@ -1,17 +1,17 @@
 host = "127.0.0.1"
 port = 8080
-path = "data2/"
+path = "data/"
 writer_memory = 200000000
 log_level = "info"
 json_parsing_threads = 4
 bulk_buffer_size = 10000
 auto_commit_duration = 10
-experimental = true
+experimental = false
 
 [experimental_features]
 master = true
 nodes = [
-    "[::1]:8081"
+    "127.0.0.1:8081"
 ]
 
 [merge_policy]

config/config.toml

@@ -128,11 +128,11 @@ fn run_master(catalog: Arc<RwLock<IndexCatalog>>, settings: &Settings) -> impl F
                 let update = catalog.read().unwrap().update_remote_indexes();
                 tokio::spawn(update);
             }
-            router_with_catalog(&bind, &catalog)
+            router_with_catalog(&bind, Arc::clone(&catalog))
         });
         future::Either::A(run)
     } else {
-        let run = commit_watcher.and_then(move |_| router_with_catalog(&bind, &catalog));
+        let run = commit_watcher.and_then(move |_| router_with_catalog(&bind, Arc::clone(&catalog)));
         future::Either::B(run)
     }
 }

src/bin/toshi.rs

@@ -7,8 +7,8 @@ use hyper::body::Body;
 use hyper::client::HttpConnector;
 use hyper::http::uri::Scheme;
 use hyper::{Client, Request, Response, Uri};
-use hyper_tls::HttpsConnector;
 use serde::{Deserialize, Serialize};
+use tower::Service;
 use tower_consul::{Consul as TowerConsul, ConsulService, KVValue};
 
 use crate::cluster::shard::PrimaryShard;
@@ -203,19 +203,19 @@ impl Builder {
 
 #[derive(Clone)]
 pub struct HttpsService {
-    client: Client<HttpsConnector<HttpConnector>>,
+    client: Client<HttpConnector>,
 }
 
 impl HttpsService {
     fn new() -> Self {
-        let https = HttpsConnector::new(4).expect("Could not create TLS for Hyper");
+        let https = HttpConnector::new(4);
         let client = Client::builder().build::<_, hyper::Body>(https);
 
         HttpsService { client }
     }
 }
 
-impl tower_service::Service<Request<Bytes>> for HttpsService {
+impl Service<Request<Bytes>> for HttpsService {
     type Response = Response<Bytes>;
     type Error = hyper::Error;
     type Future = Box<Future<Item = Self::Response, Error = Self::Error> + Send>;

src/cluster/consul.rs

@@ -3,12 +3,10 @@ use std::net::SocketAddr;
 use std::time::Duration;
 
 use failure::Fail;
-use futures::{future, Future, Poll};
+use futures::{future, Future};
 use log::error;
 use serde::{Deserialize, Serialize};
-use tokio::net::tcp::ConnectFuture;
-use tokio::net::TcpStream;
-use tower_h2::client::ConnectError;
+use tower_hyper::client::ConnectError;
 
 use crate::cluster::consul::Hosts;
 use crate::settings::Settings;
@@ -150,20 +148,3 @@ impl From<tower_grpc::Status> for RPCError {
         RPCError::RPCError(err)
     }
 }
-
-#[derive(Debug, Clone)]
-pub struct GrpcConn(pub SocketAddr);
-
-impl tower_service::Service<()> for GrpcConn {
-    type Response = TcpStream;
-    type Error = io::Error;
-    type Future = ConnectFuture;
-
-    fn poll_ready(&mut self) -> Poll<(), Self::Error> {
-        Ok(().into())
-    }
-
-    fn call(&mut self, _: ()) -> Self::Future {
-        TcpStream::connect(&self.0)
-    }
-}

src/cluster/mod.rs

@@ -6,13 +6,11 @@ use futures_watch::Watch;
 use log::error;
 use tokio::net::TcpListener;
 use tower_grpc::{Request, Response, Status};
-use tower_h2::Server;
 
 use crate::cluster::consul::Consul;
 use toshi_proto::placement_proto::{server, PlacementReply, PlacementRequest};
 
 pub use self::background::Background;
-use tokio_executor::DefaultExecutor;
 
 pub mod background;
 
@@ -35,14 +33,14 @@ impl Place {
             let placer = Place { consul, nodes };
             let placement = server::PlacementServer::new(placer);
 
-            //            let mut hyp = tower_hyper::server::Server::new(placement);
-            let mut h2 = Server::new(placement, Default::default(), DefaultExecutor::current());
+            let mut hyp = tower_hyper::server::Server::new(placement);
+
             bind.incoming().for_each(move |stream| {
                 if let Err(e) = stream.set_nodelay(true) {
                     return Err(e);
                 }
 
-                let serve = h2.serve(stream);
+                let serve = hyp.serve(stream);
                 tokio::spawn(serve.map_err(|e| error!("Placement Server Error: {:?}", e)));
 
                 Ok(())

src/cluster/placement/mod.rs

@@ -5,7 +5,7 @@ use rand::prelude::*;
 use tokio::prelude::*;
 use tower_grpc::{Request as TowerRequest, Response};
 
-use toshi_proto::cluster_rpc::{DocumentRequest, ResultReply, SearchReply, SearchRequest};
+use toshi_proto::cluster_rpc::{DeleteRequest, DocumentRequest, SearchReply, SearchRequest};
 
 use crate::cluster::rpc_server::RpcClient;
 use crate::cluster::RPCError;
@@ -49,7 +49,7 @@ impl RemoteIndex {
 
 impl IndexHandle for RemoteIndex {
     type SearchResponse = Box<Future<Item = Vec<SearchReply>, Error = RPCError> + Send>;
-    type DeleteResponse = Box<Future<Item = ResultReply, Error = RPCError> + Send>;
+    type DeleteResponse = Box<Future<Item = Vec<i32>, Error = RPCError> + Send>;
     type AddResponse = Box<Future<Item = Vec<i32>, Error = RPCError> + Send>;
 
     fn get_name(&self) -> String {
@@ -93,7 +93,7 @@ impl IndexHandle for RemoteIndex {
                 Err(_) => Vec::new(),
             };
             let req = TowerRequest::new(DocumentRequest {
-                index: name.clone(),
+                index: name,
                 document: bytes,
             });
             client
@@ -111,7 +111,30 @@ impl IndexHandle for RemoteIndex {
         Box::new(future::join_all(fut))
     }
 
-    fn delete_term(&mut self, _: DeleteDoc) -> Self::DeleteResponse {
-        unimplemented!()
+    fn delete_term(&self, delete: DeleteDoc) -> Self::DeleteResponse {
+        let name = self.name.clone();
+        let clients = self.remotes.clone();
+        let fut = clients.into_iter().map(move |mut client| {
+            let bytes = match serde_json::to_vec(&delete) {
+                Ok(v) => v,
+                Err(_) => Vec::new(),
+            };
+            let req = TowerRequest::new(DeleteRequest {
+                index: name.clone(),
+                terms: bytes,
+            });
+            client
+                .delete_document(req)
+                .map(|res| {
+                    info!("RESPONSE = {:?}", res);
+                    res.into_inner().code
+                })
+                .map_err(|e| {
+                    info!("ERR = {:?}", e);
+                    e.into()
+                })
+        });
+
+        Box::new(future::join_all(fut))
     }
 }