Use encode_varint for vector size

infinyon · Feb 9, 2021 · 66d7e4e · 66d7e4e
1 parent a533495
commit 66d7e4e
Show file tree

Hide file tree

Showing 22 changed files with 96 additions and 84 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-0.7.0-alpha.4
+0.7.0-alpha.5
diff --git a/src/auth/Cargo.toml b/src/auth/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-auth"
-version = "0.3.0"
+version = "0.4.0"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 edition = "2018"
 license = "Apache-2.0"
@@ -14,8 +14,8 @@ path = "src/lib.rs"
 
 [dependencies]
 async-trait = "0.1.41"
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 fluvio-future = { version = "0.1.12", features = ["net", "openssl_tls"] }
 fluvio-protocol = { path = "../protocol",  version = "0.3.0" }
 fluvio-socket = { path = "../socket", version = "0.5.0" }

diff --git a/src/cli/Cargo.toml b/src/cli/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-cli"
-version = "0.4.1"
+version = "0.5.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio CLI"
@@ -48,12 +48,12 @@ sysinfo = "0.16.1"
 k8-config = { version = "1.3.0" }
 k8-client = { version = "5.0.0" }
 fluvio-future = { version = "0.1.8", features = ["fs", "io", "subscriber"] }
-fluvio = { version = "0.4.1", path = "../client", default-features = false }
+fluvio = { version = "0.5.0", path = "../client", default-features = false }
 fluvio-cluster = { version = "0.7.0", path = "../cluster", default-features = false, features = ["cli"] }
 fluvio-package-index = { version = "0.2.0", path = "../package-index" }
 fluvio-extension-consumer = { version = "0.2.0", path = "../extension-consumer" }
-fluvio-extension-common = { version = "0.1.0", path = "../extension-common", features = ["target"]}
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata", features = ["use_serde", "k8"] }
+fluvio-extension-common = { version = "0.2.0", path = "../extension-common", features = ["target"]}
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata", features = ["use_serde", "k8"] }
 k8-types = { version = "0.1.0", features = ["core"]}
 
 [dev-dependencies]

diff --git a/src/client/Cargo.toml b/src/client/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio"
-version = "0.4.1"
+version = "0.5.0"
 edition = "2018"
 license = "Apache-2.0"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
@@ -38,11 +38,11 @@ semver = "0.11.0"
 # Fluvio dependencies
 fluvio-future = { version = "0.1.15", features = ["task", "native2_tls"] }
 fluvio-types = { version = "0.2.0", path = "../types" }
-fluvio-sc-schema = { version = "0.5.0", path = "../sc-schema", default-features = false }
-fluvio-spu-schema = { version = "0.3.0", path = "../spu-schema" }
+fluvio-sc-schema = { version = "0.6.0", path = "../sc-schema", default-features = false }
+fluvio-spu-schema = { version = "0.4.0", path = "../spu-schema" }
 fluvio-socket = { path = "../socket", version = "0.5.0", features = ["tls"] }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 
 [dev-dependencies]
 async-std = "1.6.4"
diff --git a/src/client/src/lib.rs b/src/client/src/lib.rs
@@ -102,7 +102,7 @@ pub use crate::admin::FluvioAdmin;
 pub use crate::client::Fluvio;
 
 /// The minimum VERSION of the Fluvio Platform that this client is compatible with.
-const MINIMUM_PLATFORM_VERSION: &str = "0.7.0-alpha.1";
+const MINIMUM_PLATFORM_VERSION: &str = "0.7.0-alpha.5";
 
 /// Creates a producer that sends events to the named topic
 ///

diff --git a/src/cluster/Cargo.toml b/src/cluster/Cargo.toml
@@ -42,13 +42,13 @@ proclist = "0.9.2"
 remoteprocess = "0.4.2"
 
 # Fluvio dependencies
-fluvio = { version = "0.4.1", path = "../client", default-features = false }
+fluvio = { version = "0.5.0", path = "../client", default-features = false }
 fluvio-helm = "0.4.1"
 fluvio-future = { version = "0.1.13" }
 fluvio-command = { version = "0.2.0", path = "../command" }
-fluvio-runner-local = { version = "0.2.0", path = "../extension-runner-local", optional = true }
-fluvio-extension-common = { version = "0.1.0", path = "../extension-common", optional = true }
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata", features = ["k8"] }
+fluvio-runner-local = { version = "0.3.0", path = "../extension-runner-local", optional = true }
+fluvio-extension-common = { version = "0.2.0", path = "../extension-common", optional = true }
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata", features = ["k8"] }
 flv-util = "0.5.2"
 k8-config = { version = "1.3.0" }
 k8-client = { version = "5.0.0" }

diff --git a/src/controlplane-metadata/Cargo.toml b/src/controlplane-metadata/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "fluvio-controlplane-metadata"
 edition = "2018"
-version = "0.5.0"
+version = "0.6.0"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio metadata"
 repository = "https://github.com/infinyon/fluvio"
@@ -27,7 +27,7 @@ flv-util = { version = "0.5.0" }
 fluvio-types = { version = "0.2.0", path = "../types" }
 fluvio-stream-model = { path = "../stream-model", version = "0.4.0" }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
 
 
 [dev-dependencies]

diff --git a/src/controlplane/Cargo.toml b/src/controlplane/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-controlplane"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2018"
 license = "Apache-2.0"
 description = "Fluvio control plane API"
@@ -17,6 +17,6 @@ tracing = "0.1.19"
 
 # Fluvio dependencies
 fluvio-types = { path = "../types", version = "0.2.0" }
-fluvio-controlplane-metadata = { path = "../controlplane-metadata", version = "0.5.0" }
+fluvio-controlplane-metadata = { path = "../controlplane-metadata", version = "0.6.0" }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
diff --git a/src/dataplane-protocol/Cargo.toml b/src/dataplane-protocol/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-dataplane-protocol"
-version = "0.2.0"
+version = "0.3.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "data plane protocol"

diff --git a/src/dataplane-protocol/src/batch.rs b/src/dataplane-protocol/src/batch.rs
@@ -243,7 +243,7 @@ mod test {
 
         let decoded_record = batch.records.get(0).unwrap();
         println!("record crc: {}", batch.header.crc);
-        assert_eq!(batch.header.crc, 2908671645);
+        assert_eq!(batch.header.crc, 843514105);
         let b = decoded_record.value.as_ref();
         assert_eq!(b, b"test");
 

diff --git a/src/dataplane-protocol/src/record.rs b/src/dataplane-protocol/src/record.rs
@@ -10,8 +10,8 @@ use log::{trace, warn};
 use once_cell::sync::Lazy;
 
 use crate::core::bytes::Buf;
-use crate::core::bytes::BufExt;
 use crate::core::bytes::BufMut;
+use bytes::buf::BufExt;
 
 use crate::core::Decoder;
 use crate::core::DecoderVarInt;
@@ -138,14 +138,22 @@ impl<'a> From<&'a [u8]> for DefaultAsyncBuffer {
 
 impl Encoder for DefaultAsyncBuffer {
     fn write_size(&self, version: Version) -> usize {
-        self.0.write_size(version)
+        let len = self.0.len() as i64;
+        self.0.iter().fold(len.var_write_size(), |sum, val| {
+            sum + val.write_size(version)
+        })
     }
 
-    fn encode<T>(&self, src: &mut T, version: Version) -> Result<(), Error>
+    fn encode<T>(&self, dest: &mut T, version: Version) -> Result<(), Error>
     where
         T: BufMut,
     {
-        self.0.encode(src, version)
+        let len: i64 = self.0.len() as i64;
+        len.encode_varint(dest)?;
+        for v in self.0.iter() {
+            v.encode(dest, version)?;
+        }
+        Ok(())
     }
 }
 
@@ -156,7 +164,14 @@ impl Decoder for DefaultAsyncBuffer {
     {
         trace!("decoding default asyncbuffer");
         if let Some(ref mut buffer) = Arc::get_mut(&mut self.0) {
-            buffer.decode(src, version)
+            let mut len: i64 = 0;
+            len.decode_varint(src)?;
+            for _ in 0..len {
+                let mut value = <u8>::default();
+                value.decode(src, version)?;
+                buffer.push(value);
+            }
+            Ok(())
         } else {
             Err(Error::new(
                 ErrorKind::Other,
@@ -434,10 +449,7 @@ mod test {
         record.encode(&mut out, 0)?;
         println!("ENCODED: {:#x?}", out);
         */
-
-        let data = [
-            0x1e, 0x0, 0x0, 0x2, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x3, 0x64, 0x6f, 0x67, 0x0,
-        ];
+        let data = [0x12, 0x0, 0x0, 0x2, 0x0, 0x6, 0x64, 0x6f, 0x67, 0x0];
 
         let record = DefaultRecord::decode_from(&mut Cursor::new(&data), 0)?;
         assert_eq!(record.as_bytes(0)?.len(), data.len());

diff --git a/src/extension-common/Cargo.toml b/src/extension-common/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-extension-common"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio extension common"
@@ -28,4 +28,4 @@ futures-lite = { version = "1.7.0" }
 thiserror = "1.0.20"
 
 
-fluvio = { version = "0.4.0", path = "../client",  optional = true }
+fluvio = { version = "0.5.0", path = "../client",  optional = true }
diff --git a/src/extension-consumer/Cargo.toml b/src/extension-consumer/Cargo.toml
@@ -30,8 +30,8 @@ hostname-validator = "1.0.0"
 
 flv-util = { version = "0.5.0" }
 fluvio-future = { version = "0.1.8", features = ["fs", "io"] }
-fluvio = { version = "0.4.1", path = "../client", default-features = false }
+fluvio = { version = "0.5.0", path = "../client", default-features = false }
 fluvio-types = { version = "0.2.0" , path = "../types" }
-fluvio-extension-common = { version = "0.1.0", path = "../extension-common", features = ["target"] }
-fluvio-controlplane-metadata = { version = "0.5.0", path = "../controlplane-metadata", features = ["use_serde"] }
-fluvio-sc-schema = { version = "0.5.0", path = "../sc-schema", features = ["use_serde"] }
+fluvio-extension-common = { version = "0.2.0", path = "../extension-common", features = ["target"] }
+fluvio-controlplane-metadata = { version = "0.6.0", path = "../controlplane-metadata", features = ["use_serde"] }
+fluvio-sc-schema = { version = "0.6.0", path = "../sc-schema", features = ["use_serde"] }
diff --git a/src/extension-runner-local/Cargo.toml b/src/extension-runner-local/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-runner-local"
-version = "0.2.0"
+version = "0.3.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio Engine Runner"
@@ -22,5 +22,5 @@ thiserror = "1.0.20"
 
 # regardless of TLS, sc and spu always use openssl_tls for now because we need cert API
 fluvio-future = { version = "0.1.12", features = ["subscriber"] }
-fluvio-sc = { version = "0.4.0", path = "../sc" }
-fluvio-spu = { version = "0.3.0", path = "../spu" }
+fluvio-sc = { version = "0.5.0", path = "../sc" }
+fluvio-spu = { version = "0.4.0", path = "../spu" }
diff --git a/src/sc-schema/Cargo.toml b/src/sc-schema/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fluvio-sc-schema"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 description = "Fluvio API for SC"
@@ -23,6 +23,6 @@ static_assertions = "1.1.0"
 
 # Fluvio dependencies
 fluvio-types = { version = "0.2.0", path = "../types" }
-fluvio-controlplane-metadata = { version = "0.5.0", default-features = false, path = "../controlplane-metadata" }
+fluvio-controlplane-metadata = { version = "0.6.0", default-features = false, path = "../controlplane-metadata" }
 fluvio-protocol = { path = "../protocol", version = "0.3.0" }
-dataplane = { version = "0.2.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }
+dataplane = { version = "0.3.0", path = "../dataplane-protocol", package = "fluvio-dataplane-protocol" }