fix transaction api v1 exmaple

Author: iosmanthus

examples/transaction.rs

@@ -3,9 +3,15 @@
 mod common;
 
 use crate::common::parse_args;
-use tikv_client::{BoundRange, Config, Key, KvPair, TransactionClient as Client, Value};
+use tikv_client::{
+    request::request_codec::{RequestCodec, TxnApiV1},
+    BoundRange, Config, Key, KvPair, TransactionClient as Client, Value,
+};
 
-async fn puts(client: &Client, pairs: impl IntoIterator<Item = impl Into<KvPair>>) {
+async fn puts<C: RequestCodec>(
+    client: &Client<C>,
+    pairs: impl IntoIterator<Item = impl Into<KvPair>>,
+) {
     let mut txn = client
         .begin_optimistic()
         .await
@@ -17,7 +23,7 @@ async fn puts(client: &Client, pairs: impl IntoIterator<Item = impl Into<KvPair>
     txn.commit().await.expect("Could not commit transaction");
 }
 
-async fn get(client: &Client, key: Key) -> Option<Value> {
+async fn get<C: RequestCodec>(client: &Client<C>, key: Key) -> Option<Value> {
     let mut txn = client
         .begin_optimistic()
         .await
@@ -29,7 +35,7 @@ async fn get(client: &Client, key: Key) -> Option<Value> {
     res
 }
 
-async fn key_exists(client: &Client, key: Key) -> bool {
+async fn key_exists<C: RequestCodec>(client: &Client<C>, key: Key) -> bool {
     let mut txn = client
         .begin_optimistic()
         .await
@@ -44,7 +50,7 @@ async fn key_exists(client: &Client, key: Key) -> bool {
     res
 }
 
-async fn scan(client: &Client, range: impl Into<BoundRange>, limit: u32) {
+async fn scan<C: RequestCodec>(client: &Client<C>, range: impl Into<BoundRange>, limit: u32) {
     let mut txn = client
         .begin_optimistic()
         .await
@@ -56,7 +62,7 @@ async fn scan(client: &Client, range: impl Into<BoundRange>, limit: u32) {
     txn.commit().await.expect("Could not commit transaction");
 }
 
-async fn dels(client: &Client, keys: impl IntoIterator<Item = Key>) {
+async fn dels<C: RequestCodec>(client: &Client<C>, keys: impl IntoIterator<Item = Key>) {
     let mut txn = client
         .begin_optimistic()
         .await
@@ -81,7 +87,7 @@ async fn main() {
         Config::default()
     };
 
-    let txn = Client::new_with_config(args.pd, config, None)
+    let txn = Client::new_with_config(args.pd, config, TxnApiV1, None)
         .await
         .expect("Could not connect to tikv");
 

src/lib.rs

@@ -90,8 +90,7 @@
 //! # })}
 //! ```
 
-#![allow(incomplete_features)]
-#![feature(specialization)]
+#![feature(min_specialization)]
 #![feature(explicit_generic_args_with_impl_trait)]
 #[macro_use]
 pub mod request;

src/pd/client.rs

@@ -14,7 +14,6 @@ use tikv_client_store::{KvClient, KvConnect, TikvConnect};
 
 use crate::{
     compat::stream_fn,
-    kv::codec,
     pd::{retry::RetryClientTrait, RetryClient},
     region::{RegionId, RegionVerId, RegionWithLeader},
     region_cache::RegionCache,
@@ -198,14 +197,6 @@ pub trait PdClient: Send + Sync + 'static {
         .boxed()
     }
 
-    fn decode_region(mut region: RegionWithLeader, enable_codec: bool) -> Result<RegionWithLeader> {
-        if enable_codec {
-            codec::decode_bytes_in_place(region.region.mut_start_key(), false)?;
-            codec::decode_bytes_in_place(region.region.mut_end_key(), false)?;
-        }
-        Ok(region)
-    }
-
     async fn update_leader(&self, ver_id: RegionVerId, leader: metapb::Peer) -> Result<()>;
 
     async fn invalidate_region_cache(&self, ver_id: RegionVerId);

src/region_cache.rs

@@ -132,14 +132,14 @@ impl<C: RequestCodec, R: RetryClientTrait> RegionCache<C, R> {
 
     /// Force read through (query from PD) and update cache
     pub async fn read_through_region_by_key(&self, key: Key) -> Result<RegionWithLeader> {
-        let mut region = self
+        let mut r = self
             .inner_client
             .clone()
             .get_region(self.codec.encode_pd_query(key).into())
             .await?;
-        region.region = self.codec.decode_region(region.region)?;
-        self.add_region(region.clone()).await;
-        Ok(region)
+        r.region = self.codec.decode_region(r.region)?;
+        self.add_region(r.clone()).await;
+        Ok(r)
     }
 
     /// Force read through (query from PD) and update cache

src/request/mod.rs

@@ -32,7 +32,7 @@ pub mod request_codec;
 #[async_trait]
 pub trait KvRequest<C>: Request + Sized + Clone + Sync + Send + 'static {
     /// The expected response to the request.
-    type Response: HasKeyErrors + HasLocks + HasRegionError + Clone + Send + 'static;
+    type Response: HasKeyErrors + HasLocks + HasRegionError + Clone + Send  + 'static;
     fn encode_request(&self, _codec: &C) -> Cow<Self>;
     fn decode_response(&self, _codec: &C, _resp: Self::Response) -> crate::Result<Self::Response>;
 }
@@ -73,7 +73,7 @@ mod test {
     use std::{
         any::Any,
         iter,
-        sync::{atomic::AtomicUsize, Arc},
+        sync::{Arc, atomic::AtomicUsize},
     };
 
     use grpcio::CallOption;
@@ -82,10 +82,10 @@ mod test {
     use tikv_client_store::HasRegionError;
 
     use crate::{
+        Error,
+        Key,
         mock::{MockKvClient, MockPdClient},
-        store::store_stream_for_keys,
-        transaction::lowering::new_commit_request,
-        Error, Key, Result,
+        Result, store::store_stream_for_keys, transaction::lowering::new_commit_request,
     };
 
     use super::*;

src/request/request_codec.rs

@@ -1,6 +1,24 @@
 use tikv_client_proto::metapb::Region;
 
-use crate::{Key, Result};
+use crate::{kv::codec::decode_bytes_in_place, Key, Result};
+
+#[macro_export]
+macro_rules! plain_request {
+    ($req: ident, $codec: ident) => {
+        if $codec.is_plain() {
+            return ::std::borrow::Cow::Borrowed($req);
+        }
+    };
+}
+
+#[macro_export]
+macro_rules! plain_response {
+    ($resp: ident, $codec: ident) => {
+        if $codec.is_plain() {
+            return Ok($resp);
+        }
+    };
+}
 
 pub trait RequestCodec: Sized + Clone + Sync + Send + 'static {
     fn encode_key(&self, key: Key) -> Key {
@@ -18,6 +36,9 @@ pub trait RequestCodec: Sized + Clone + Sync + Send + 'static {
     fn decode_region(&self, region: Region) -> Result<Region> {
         Ok(region)
     }
+    fn is_plain(&self) -> bool {
+        true
+    }
 }
 
 #[derive(Clone)]
@@ -28,4 +49,14 @@ impl RequestCodec for RawApiV1 {}
 #[derive(Clone)]
 pub struct TxnApiV1;
 
-impl RequestCodec for TxnApiV1 {}
+impl RequestCodec for TxnApiV1 {
+    fn encode_pd_query(&self, key: Key) -> Key {
+        key.to_encoded()
+    }
+
+    fn decode_region(&self, mut region: Region) -> Result<Region> {
+        decode_bytes_in_place(region.mut_start_key(), false)?;
+        decode_bytes_in_place(region.mut_end_key(), false)?;
+        Ok(region)
+    }
+}