*: upgrade grpcio to v0.4

Cargo.toml

@@ -72,7 +72,7 @@ zipf = "0.2.0"
 bitflags = "1.0.1"
 fail = "0.2"
 uuid = { version = "0.6", features = [ "serde", "v4" ] }
-grpcio = { version = "0.3", features = [ "secure" ] }
+grpcio = { version = "0.4", features = [ "secure" ] }
 raft = "0.3"
 crossbeam-channel = "0.2"
 crossbeam = "0.2"

components/test_raftstore/Cargo.toml

@@ -13,7 +13,7 @@ tempdir = "0.3"
 raft = "0.3"
 futures = "0.1"
 tokio-timer = "0.2"
-grpcio = { version = "0.3", features = [ "secure" ] }
+grpcio = { version = "0.4", features = [ "secure" ] }
 rand = "0.3"
 log = "0.3.9"
 

src/import/kv_service.rs

@@ -53,7 +53,7 @@ impl ImportKVService {
 
 impl ImportKv for ImportKVService {
     fn switch_mode(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: SwitchModeRequest,
         sink: UnarySink<SwitchModeResponse>,
@@ -82,7 +82,7 @@ impl ImportKv for ImportKVService {
     }
 
     fn open_engine(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: OpenEngineRequest,
         sink: UnarySink<OpenEngineResponse>,
@@ -103,7 +103,7 @@ impl ImportKv for ImportKVService {
     }
 
     fn write_engine(
-        &self,
+        &mut self,
         ctx: RpcContext,
         stream: RequestStream<WriteEngineRequest>,
         sink: ClientStreamingSink<WriteEngineResponse>,
@@ -160,7 +160,7 @@ impl ImportKv for ImportKVService {
     }
 
     fn close_engine(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: CloseEngineRequest,
         sink: UnarySink<CloseEngineResponse>,
@@ -191,7 +191,7 @@ impl ImportKv for ImportKVService {
     }
 
     fn import_engine(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: ImportEngineRequest,
         sink: UnarySink<ImportEngineResponse>,
@@ -212,7 +212,7 @@ impl ImportKv for ImportKVService {
     }
 
     fn cleanup_engine(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: CleanupEngineRequest,
         sink: UnarySink<CleanupEngineResponse>,
@@ -233,7 +233,7 @@ impl ImportKv for ImportKVService {
     }
 
     fn compact_cluster(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: CompactClusterRequest,
         sink: UnarySink<CompactClusterResponse>,

src/import/sst_service.rs

@@ -71,7 +71,7 @@ impl<Router: RaftStoreRouter> ImportSSTService<Router> {
 
 impl<Router: RaftStoreRouter> ImportSst for ImportSSTService<Router> {
     fn switch_mode(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: SwitchModeRequest,
         sink: UnarySink<SwitchModeResponse>,
@@ -100,7 +100,7 @@ impl<Router: RaftStoreRouter> ImportSst for ImportSSTService<Router> {
 
     /// Receive SST from client and save the file for later ingesting.
     fn upload(
-        &self,
+        &mut self,
         ctx: RpcContext,
         stream: RequestStream<UploadRequest>,
         sink: ClientStreamingSink<UploadResponse>,
@@ -154,7 +154,7 @@ impl<Router: RaftStoreRouter> ImportSst for ImportSSTService<Router> {
     /// If the ingestion fails because the region is not found or the epoch does
     /// not match, the remaining files will eventually be cleaned up by
     /// CleanupSSTWorker.
-    fn ingest(&self, ctx: RpcContext, mut req: IngestRequest, sink: UnarySink<IngestResponse>) {
+    fn ingest(&mut self, ctx: RpcContext, mut req: IngestRequest, sink: UnarySink<IngestResponse>) {
         let label = "ingest";
         let timer = Instant::now_coarse();
 
@@ -194,7 +194,7 @@ impl<Router: RaftStoreRouter> ImportSst for ImportSSTService<Router> {
         )
     }
 
-    fn compact(&self, ctx: RpcContext, req: CompactRequest, sink: UnarySink<CompactResponse>) {
+    fn compact(&mut self, ctx: RpcContext, req: CompactRequest, sink: UnarySink<CompactResponse>) {
         let label = "compact";
         let timer = Instant::now_coarse();
         let engine = Arc::clone(&self.engine);

src/server/raft_client.rs

@@ -71,7 +71,7 @@ impl Conn {
         let client = TikvClient::new(channel);
         let (tx, rx) = mpsc::unbounded();
         let (tx_close, rx_close) = oneshot::channel();
-        let (sink, _) = client.raft().unwrap();
+        let (sink, receiver) = client.raft().unwrap();
         let addr = addr.to_owned();
         client.spawn(
             rx_close
@@ -92,8 +92,7 @@ impl Conn {
                             warn!("send raftmessage to {} failed: {:?}", addr, e);
                         }),
                 )
-                .map(|_| ())
-                .map_err(|_| ()),
+                .then(|_| receiver.then(|_| Ok(()))),
         );
         Conn {
             stream: tx,

src/server/service/debug.rs

@@ -86,7 +86,7 @@ impl<T: RaftStoreRouter> Service<T> {
 }
 
 impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
-    fn get(&self, ctx: RpcContext, mut req: GetRequest, sink: UnarySink<GetResponse>) {
+    fn get(&mut self, ctx: RpcContext, mut req: GetRequest, sink: UnarySink<GetResponse>) {
         const TAG: &str = "debug_get";
 
         let db = req.get_db();
@@ -108,7 +108,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
         self.handle_response(ctx, sink, f, TAG);
     }
 
-    fn raft_log(&self, ctx: RpcContext, req: RaftLogRequest, sink: UnarySink<RaftLogResponse>) {
+    fn raft_log(&mut self, ctx: RpcContext, req: RaftLogRequest, sink: UnarySink<RaftLogResponse>) {
         const TAG: &str = "debug_raft_log";
 
         let region_id = req.get_region_id();
@@ -130,7 +130,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn region_info(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: RegionInfoRequest,
         sink: UnarySink<RegionInfoResponse>,
@@ -163,7 +163,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn region_size(
-        &self,
+        &mut self,
         ctx: RpcContext,
         mut req: RegionSizeRequest,
         sink: UnarySink<RegionSizeResponse>,
@@ -199,7 +199,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn scan_mvcc(
-        &self,
+        &mut self,
         _: RpcContext,
         mut req: ScanMvccRequest,
         sink: ServerStreamingSink<ScanMvccResponse>,
@@ -226,7 +226,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
         self.pool.spawn(future).forget();
     }
 
-    fn compact(&self, ctx: RpcContext, req: CompactRequest, sink: UnarySink<CompactResponse>) {
+    fn compact(&mut self, ctx: RpcContext, req: CompactRequest, sink: UnarySink<CompactResponse>) {
         let debugger = self.debugger.clone();
         let f = self.pool.spawn_fn(move || {
             debugger
@@ -244,7 +244,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn inject_fail_point(
-        &self,
+        &mut self,
         ctx: RpcContext,
         mut req: InjectFailPointRequest,
         sink: UnarySink<InjectFailPointResponse>,
@@ -267,7 +267,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn recover_fail_point(
-        &self,
+        &mut self,
         ctx: RpcContext,
         mut req: RecoverFailPointRequest,
         sink: UnarySink<RecoverFailPointResponse>,
@@ -287,7 +287,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn list_fail_points(
-        &self,
+        &mut self,
         ctx: RpcContext,
         _: ListFailPointsRequest,
         sink: UnarySink<ListFailPointsResponse>,
@@ -310,7 +310,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn get_metrics(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: GetMetricsRequest,
         sink: UnarySink<GetMetricsResponse>,
@@ -335,7 +335,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn check_region_consistency(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: RegionConsistencyCheckRequest,
         sink: UnarySink<RegionConsistencyCheckResponse>,
@@ -356,7 +356,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn modify_tikv_config(
-        &self,
+        &mut self,
         ctx: RpcContext,
         mut req: ModifyTikvConfigRequest,
         sink: UnarySink<ModifyTikvConfigResponse>,
@@ -378,7 +378,7 @@ impl<T: RaftStoreRouter + 'static + Send> debugpb_grpc::Debug for Service<T> {
     }
 
     fn get_region_properties(
-        &self,
+        &mut self,
         ctx: RpcContext,
         req: GetRegionPropertiesRequest,
         sink: UnarySink<GetRegionPropertiesResponse>,