You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "jaroslawroszyk (via GitHub)" <gi...@apache.org> on 2025/06/20 14:22:39 UTC
[PR] feat(connectors): Add Protocol Buffers support to connectors runtime [iggy]
jaroslawroszyk opened a new pull request, #1886:
URL: https://github.com/apache/iggy/pull/1886
- Implement ProtoStreamDecoder with configurable Any wrapper and field mappings
- Implement ProtoStreamEncoder supporting JSON, Text, and Raw payload conversion
- Add ProtoConvert transform for format conversion between Protobuf, JSON, and Text
- Support protobuf Any message wrapper with base64 encoding for binary data
- Add comprehensive unit tests for decoder, encoder, and transform components
- Extend Schema enum with Proto variant and integrate with existing decoder/encoder system
- Update dependencies to include prost and prost-types for protobuf handling
- Replace deprecated base64 encoding with new Engine API
This implementation enables connectors to work with Protocol Buffers payloads, including schema configuration, data transformations, and format conversions. The decoder supports both raw protobuf messages and Any-wrapped messages, while the encoder can convert from JSON, Text, and Raw formats to protobuf.
Future enhancements planned:
- Dynamic loading and compilation of .proto files at runtime
- Runtime code generation for protobuf message types
- Enhanced schema validation and field mapping capabilities
Closes #1845
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2159488801
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -0,0 +1,159 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use base64::Engine;
+use prost::Message;
+use prost_types::Any;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use tracing::error;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProtoConfig {
+ pub schema_path: Option<String>,
+ pub message_type: Option<String>,
+ pub use_any_wrapper: bool,
+ pub field_mappings: Option<HashMap<String, String>>,
+}
+
+impl Default for ProtoConfig {
+ fn default() -> Self {
+ Self {
+ schema_path: None,
+ message_type: None,
+ use_any_wrapper: true,
+ field_mappings: None,
+ }
+ }
+}
+
+pub struct ProtoStreamDecoder {
+ config: ProtoConfig,
+}
+
+impl ProtoStreamDecoder {
+ pub fn new(config: ProtoConfig) -> Self {
+ Self { config }
+ }
+
+ pub fn new_default() -> Self {
+ Self::new(ProtoConfig::default())
+ }
+
+ fn decode_as_any(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ match Any::decode(payload.as_slice()) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf Any message: {e}");
+ Err(Error::CannotDecode(Schema::Proto))
+ }
+ }
+ }
+
+ fn decode_as_raw(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Raw(payload))
+ }
+}
+
+impl StreamDecoder for ProtoStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Proto
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ if payload.is_empty() {
+ return Err(Error::InvalidPayloadType);
+ }
+ println!("jarek");
+ if self.config.use_any_wrapper {
+ self.decode_as_any(payload)
+ } else {
+ self.decode_as_raw(payload)
+ }
+ }
+}
+
+impl Default for ProtoStreamDecoder {
+ fn default() -> Self {
+ Self::new_default()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use prost_types::Any;
+
+ #[test]
+ fn test_decode_protobuf_any() {
+ let decoder = ProtoStreamDecoder::new_default();
+
+ let any = Any {
+ type_url: "type.googleapis.com/google.protobuf.StringValue".to_string(),
+ value: b"Hello, World!".to_vec(),
+ };
+
+ let encoded = any.encode_to_vec();
+ let result = decoder.decode(encoded);
+
+ assert!(result.is_ok());
+ if let Ok(Payload::Json(json_value)) = result {
+ if let simd_json::OwnedValue::Object(map) = &json_value {
+ assert!(map.contains_key("type_url"));
+ assert!(map.contains_key("value"));
+ } else {
+ panic!("Expected JSON object");
+ }
+ } else {
+ panic!("Expected JSON payload");
+ }
+ }
+
+ #[test]
+ fn test_decode_empty_payload() {
Review Comment:
Let's name the test methods by following the basic BDD convention e.g. `decode_should_fail_given_empty_payload` or something like this :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on PR #1886:
URL: https://github.com/apache/iggy/pull/1886#issuecomment-3001815462
@hubcio Should be done until Thursday
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2173282842
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -0,0 +1,159 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use base64::Engine;
+use prost::Message;
+use prost_types::Any;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use tracing::error;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProtoConfig {
+ pub schema_path: Option<String>,
+ pub message_type: Option<String>,
+ pub use_any_wrapper: bool,
+ pub field_mappings: Option<HashMap<String, String>>,
+}
+
+impl Default for ProtoConfig {
+ fn default() -> Self {
+ Self {
+ schema_path: None,
+ message_type: None,
+ use_any_wrapper: true,
+ field_mappings: None,
+ }
+ }
+}
+
+pub struct ProtoStreamDecoder {
+ config: ProtoConfig,
+}
+
+impl ProtoStreamDecoder {
+ pub fn new(config: ProtoConfig) -> Self {
+ Self { config }
+ }
+
+ pub fn new_default() -> Self {
+ Self::new(ProtoConfig::default())
+ }
+
+ fn decode_as_any(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ match Any::decode(payload.as_slice()) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf Any message: {e}");
+ Err(Error::CannotDecode(Schema::Proto))
+ }
+ }
+ }
+
+ fn decode_as_raw(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Raw(payload))
+ }
+}
+
+impl StreamDecoder for ProtoStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Proto
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ if payload.is_empty() {
+ return Err(Error::InvalidPayloadType);
+ }
+ println!("jarek");
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2173282001
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -145,6 +169,7 @@ impl Schema {
Schema::Json => Arc::new(JsonStreamEncoder),
Schema::Raw => Arc::new(RawStreamEncoder),
Schema::Text => Arc::new(TextStreamEncoder),
+ Schema::Proto => Arc::new(ProtoStreamEncoder::new()),
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on PR #1886:
URL: https://github.com/apache/iggy/pull/1886#issuecomment-3016623954
Looks good, please update the licenses check and it should be g2g :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2159487210
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -0,0 +1,159 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use base64::Engine;
+use prost::Message;
+use prost_types::Any;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use tracing::error;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProtoConfig {
+ pub schema_path: Option<String>,
+ pub message_type: Option<String>,
+ pub use_any_wrapper: bool,
+ pub field_mappings: Option<HashMap<String, String>>,
+}
+
+impl Default for ProtoConfig {
+ fn default() -> Self {
+ Self {
+ schema_path: None,
+ message_type: None,
+ use_any_wrapper: true,
+ field_mappings: None,
+ }
+ }
+}
+
+pub struct ProtoStreamDecoder {
+ config: ProtoConfig,
+}
+
+impl ProtoStreamDecoder {
+ pub fn new(config: ProtoConfig) -> Self {
+ Self { config }
+ }
+
+ pub fn new_default() -> Self {
+ Self::new(ProtoConfig::default())
+ }
+
+ fn decode_as_any(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ match Any::decode(payload.as_slice()) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf Any message: {e}");
+ Err(Error::CannotDecode(Schema::Proto))
+ }
+ }
+ }
+
+ fn decode_as_raw(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Raw(payload))
+ }
+}
+
+impl StreamDecoder for ProtoStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Proto
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ if payload.is_empty() {
+ return Err(Error::InvalidPayloadType);
+ }
+ println!("jarek");
Review Comment:
This line probably can be removed :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2173283347
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -0,0 +1,159 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use base64::Engine;
+use prost::Message;
+use prost_types::Any;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use tracing::error;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProtoConfig {
+ pub schema_path: Option<String>,
+ pub message_type: Option<String>,
+ pub use_any_wrapper: bool,
+ pub field_mappings: Option<HashMap<String, String>>,
+}
+
+impl Default for ProtoConfig {
+ fn default() -> Self {
+ Self {
+ schema_path: None,
+ message_type: None,
+ use_any_wrapper: true,
+ field_mappings: None,
+ }
+ }
+}
+
+pub struct ProtoStreamDecoder {
+ config: ProtoConfig,
+}
+
+impl ProtoStreamDecoder {
+ pub fn new(config: ProtoConfig) -> Self {
+ Self { config }
+ }
+
+ pub fn new_default() -> Self {
+ Self::new(ProtoConfig::default())
+ }
+
+ fn decode_as_any(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ match Any::decode(payload.as_slice()) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf Any message: {e}");
+ Err(Error::CannotDecode(Schema::Proto))
+ }
+ }
+ }
+
+ fn decode_as_raw(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Raw(payload))
+ }
+}
+
+impl StreamDecoder for ProtoStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Proto
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ if payload.is_empty() {
+ return Err(Error::InvalidPayloadType);
+ }
+ println!("jarek");
+ if self.config.use_any_wrapper {
+ self.decode_as_any(payload)
+ } else {
+ self.decode_as_raw(payload)
+ }
+ }
+}
+
+impl Default for ProtoStreamDecoder {
+ fn default() -> Self {
+ Self::new_default()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use prost_types::Any;
+
+ #[test]
+ fn test_decode_protobuf_any() {
+ let decoder = ProtoStreamDecoder::new_default();
+
+ let any = Any {
+ type_url: "type.googleapis.com/google.protobuf.StringValue".to_string(),
+ value: b"Hello, World!".to_vec(),
+ };
+
+ let encoded = any.encode_to_vec();
+ let result = decoder.decode(encoded);
+
+ assert!(result.is_ok());
+ if let Ok(Payload::Json(json_value)) = result {
+ if let simd_json::OwnedValue::Object(map) = &json_value {
+ assert!(map.contains_key("type_url"));
+ assert!(map.contains_key("value"));
+ } else {
+ panic!("Expected JSON object");
+ }
+ } else {
+ panic!("Expected JSON payload");
+ }
+ }
+
+ #[test]
+ fn test_decode_empty_payload() {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio merged PR #1886:
URL: https://github.com/apache/iggy/pull/1886
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2173281126
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -0,0 +1,159 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use base64::Engine;
+use prost::Message;
+use prost_types::Any;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use tracing::error;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProtoConfig {
+ pub schema_path: Option<String>,
+ pub message_type: Option<String>,
+ pub use_any_wrapper: bool,
+ pub field_mappings: Option<HashMap<String, String>>,
+}
+
+impl Default for ProtoConfig {
+ fn default() -> Self {
+ Self {
+ schema_path: None,
+ message_type: None,
+ use_any_wrapper: true,
+ field_mappings: None,
+ }
+ }
+}
+
+pub struct ProtoStreamDecoder {
+ config: ProtoConfig,
+}
+
+impl ProtoStreamDecoder {
+ pub fn new(config: ProtoConfig) -> Self {
+ Self { config }
+ }
+
+ pub fn new_default() -> Self {
+ Self::new(ProtoConfig::default())
+ }
+
+ fn decode_as_any(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ match Any::decode(payload.as_slice()) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf Any message: {e}");
+ Err(Error::CannotDecode(Schema::Proto))
+ }
+ }
+ }
+
+ fn decode_as_raw(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Raw(payload))
+ }
+}
+
+impl StreamDecoder for ProtoStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Proto
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ if payload.is_empty() {
+ return Err(Error::InvalidPayloadType);
+ }
+ println!("jarek");
+ if self.config.use_any_wrapper {
+ self.decode_as_any(payload)
+ } else {
+ self.decode_as_raw(payload)
+ }
+ }
+}
+
+impl Default for ProtoStreamDecoder {
+ fn default() -> Self {
+ Self::new_default()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use prost_types::Any;
+
+ #[test]
+ fn test_decode_protobuf_any() {
+ let decoder = ProtoStreamDecoder::new_default();
Review Comment:
Done
##########
core/connectors/sdk/examples/protobuf_config.toml:
##########
@@ -0,0 +1,43 @@
+[iggy]
+address = "localhost:8090"
+username = "iggy"
+password = "iggy"
+
+[sources.protobuf_source]
Review Comment:
Done
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -137,6 +160,7 @@ impl Schema {
Schema::Json => Arc::new(JsonStreamDecoder),
Schema::Raw => Arc::new(RawStreamDecoder),
Schema::Text => Arc::new(TextStreamDecoder),
+ Schema::Proto => Arc::new(ProtoStreamDecoder::new_default()),
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2159481389
##########
core/connectors/sdk/src/decoders/proto.rs:
##########
@@ -0,0 +1,159 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use crate::{Error, Payload, Schema, StreamDecoder};
+use base64::Engine;
+use prost::Message;
+use prost_types::Any;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use tracing::error;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ProtoConfig {
+ pub schema_path: Option<String>,
+ pub message_type: Option<String>,
+ pub use_any_wrapper: bool,
+ pub field_mappings: Option<HashMap<String, String>>,
+}
+
+impl Default for ProtoConfig {
+ fn default() -> Self {
+ Self {
+ schema_path: None,
+ message_type: None,
+ use_any_wrapper: true,
+ field_mappings: None,
+ }
+ }
+}
+
+pub struct ProtoStreamDecoder {
+ config: ProtoConfig,
+}
+
+impl ProtoStreamDecoder {
+ pub fn new(config: ProtoConfig) -> Self {
+ Self { config }
+ }
+
+ pub fn new_default() -> Self {
+ Self::new(ProtoConfig::default())
+ }
+
+ fn decode_as_any(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ match Any::decode(payload.as_slice()) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf Any message: {e}");
+ Err(Error::CannotDecode(Schema::Proto))
+ }
+ }
+ }
+
+ fn decode_as_raw(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ Ok(Payload::Raw(payload))
+ }
+}
+
+impl StreamDecoder for ProtoStreamDecoder {
+ fn schema(&self) -> Schema {
+ Schema::Proto
+ }
+
+ fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
+ if payload.is_empty() {
+ return Err(Error::InvalidPayloadType);
+ }
+ println!("jarek");
+ if self.config.use_any_wrapper {
+ self.decode_as_any(payload)
+ } else {
+ self.decode_as_raw(payload)
+ }
+ }
+}
+
+impl Default for ProtoStreamDecoder {
+ fn default() -> Self {
+ Self::new_default()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use prost_types::Any;
+
+ #[test]
+ fn test_decode_protobuf_any() {
+ let decoder = ProtoStreamDecoder::new_default();
Review Comment:
Maybe let's just call `default` here?
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -137,6 +160,7 @@ impl Schema {
Schema::Json => Arc::new(JsonStreamDecoder),
Schema::Raw => Arc::new(RawStreamDecoder),
Schema::Text => Arc::new(TextStreamDecoder),
+ Schema::Proto => Arc::new(ProtoStreamDecoder::new_default()),
Review Comment:
Also here, just a `default`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #1886:
URL: https://github.com/apache/iggy/pull/1886#issuecomment-3001674343
@jaroslawroszyk any plans to continue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2173282486
##########
core/connectors/sdk/src/transforms/proto_convert.rs:
##########
@@ -0,0 +1,143 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use prost::Message;
+use prost_types::Any;
+use base64::Engine;
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
+use tracing::error;
+
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
+
+use super::{Transform, TransformType};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ProtoConvertConfig {
+ pub target_format: ConvertFormat,
+ pub preserve_structure: bool,
+ pub field_mappings: Option<std::collections::HashMap<String, String>>,
+}
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ConvertFormat {
+ Json,
+ Proto,
+ Text,
+}
+
+pub struct ProtoConvert {
+ config: ProtoConvertConfig,
+}
+
+impl ProtoConvert {
+ pub fn new(config: ProtoConvertConfig) -> Self {
+ Self { config }
+ }
+
+ fn convert_proto_to_json(&self, payload: &[u8]) -> Result<Payload, Error> {
+ match Any::decode(payload) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ "converted_at": chrono::Utc::now().to_rfc3339(),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf message: {e}");
+ Err(Error::InvalidProtobufPayload)
+ }
+ }
+ }
+
+ fn convert_json_to_proto(&self, json_value: &OwnedValue) -> Result<Payload, Error> {
+ let json_string = simd_json::to_string(json_value)
+ .map_err(|_| Error::InvalidJsonPayload)?;
+
+ let any = Any {
+ type_url: "type.googleapis.com/google.protobuf.StringValue".to_string(),
+ value: json_string.into_bytes(),
+ };
+
+ let proto_bytes = any.encode_to_vec();
+ Ok(Payload::Raw(proto_bytes))
+ }
+
+ fn convert_to_text(&self, payload: &Payload) -> Result<Payload, Error> {
+ match payload {
+ Payload::Json(value) => {
+ let text = simd_json::to_string_pretty(value)
+ .map_err(|_| Error::InvalidJsonPayload)?;
+ Ok(Payload::Text(text))
+ }
+ Payload::Proto(text) => Ok(Payload::Text(text.clone())),
+ Payload::Raw(data) => {
+ match self.convert_proto_to_json(data) {
+ Ok(Payload::Json(value)) => {
+ let text = simd_json::to_string_pretty(&value)
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2159483869
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -145,6 +169,7 @@ impl Schema {
Schema::Json => Arc::new(JsonStreamEncoder),
Schema::Raw => Arc::new(RawStreamEncoder),
Schema::Text => Arc::new(TextStreamEncoder),
+ Schema::Proto => Arc::new(ProtoStreamEncoder::new()),
Review Comment:
If it's an empty struct, could we just pass `ProtoStreamEncoder`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #1886:
URL: https://github.com/apache/iggy/pull/1886#issuecomment-3001828017
@jaroslawroszyk ping
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2159486232
##########
core/connectors/sdk/src/transforms/proto_convert.rs:
##########
@@ -0,0 +1,143 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use prost::Message;
+use prost_types::Any;
+use base64::Engine;
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
+use tracing::error;
+
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
+
+use super::{Transform, TransformType};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ProtoConvertConfig {
+ pub target_format: ConvertFormat,
+ pub preserve_structure: bool,
+ pub field_mappings: Option<std::collections::HashMap<String, String>>,
+}
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ConvertFormat {
+ Json,
+ Proto,
+ Text,
+}
+
+pub struct ProtoConvert {
+ config: ProtoConvertConfig,
+}
+
+impl ProtoConvert {
+ pub fn new(config: ProtoConvertConfig) -> Self {
+ Self { config }
+ }
+
+ fn convert_proto_to_json(&self, payload: &[u8]) -> Result<Payload, Error> {
+ match Any::decode(payload) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ "converted_at": chrono::Utc::now().to_rfc3339(),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf message: {e}");
+ Err(Error::InvalidProtobufPayload)
+ }
+ }
+ }
+
+ fn convert_json_to_proto(&self, json_value: &OwnedValue) -> Result<Payload, Error> {
+ let json_string = simd_json::to_string(json_value)
+ .map_err(|_| Error::InvalidJsonPayload)?;
+
+ let any = Any {
+ type_url: "type.googleapis.com/google.protobuf.StringValue".to_string(),
+ value: json_string.into_bytes(),
+ };
+
+ let proto_bytes = any.encode_to_vec();
+ Ok(Payload::Raw(proto_bytes))
+ }
+
+ fn convert_to_text(&self, payload: &Payload) -> Result<Payload, Error> {
+ match payload {
+ Payload::Json(value) => {
+ let text = simd_json::to_string_pretty(value)
+ .map_err(|_| Error::InvalidJsonPayload)?;
+ Ok(Payload::Text(text))
+ }
+ Payload::Proto(text) => Ok(Payload::Text(text.clone())),
+ Payload::Raw(data) => {
+ match self.convert_proto_to_json(data) {
+ Ok(Payload::Json(value)) => {
+ let text = simd_json::to_string_pretty(&value)
Review Comment:
And also here, just `to_string()` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2159485881
##########
core/connectors/sdk/src/transforms/proto_convert.rs:
##########
@@ -0,0 +1,143 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use prost::Message;
+use prost_types::Any;
+use base64::Engine;
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
+use tracing::error;
+
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
+
+use super::{Transform, TransformType};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ProtoConvertConfig {
+ pub target_format: ConvertFormat,
+ pub preserve_structure: bool,
+ pub field_mappings: Option<std::collections::HashMap<String, String>>,
+}
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ConvertFormat {
+ Json,
+ Proto,
+ Text,
+}
+
+pub struct ProtoConvert {
+ config: ProtoConvertConfig,
+}
+
+impl ProtoConvert {
+ pub fn new(config: ProtoConvertConfig) -> Self {
+ Self { config }
+ }
+
+ fn convert_proto_to_json(&self, payload: &[u8]) -> Result<Payload, Error> {
+ match Any::decode(payload) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ "converted_at": chrono::Utc::now().to_rfc3339(),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf message: {e}");
+ Err(Error::InvalidProtobufPayload)
+ }
+ }
+ }
+
+ fn convert_json_to_proto(&self, json_value: &OwnedValue) -> Result<Payload, Error> {
+ let json_string = simd_json::to_string(json_value)
+ .map_err(|_| Error::InvalidJsonPayload)?;
+
+ let any = Any {
+ type_url: "type.googleapis.com/google.protobuf.StringValue".to_string(),
+ value: json_string.into_bytes(),
+ };
+
+ let proto_bytes = any.encode_to_vec();
+ Ok(Payload::Raw(proto_bytes))
+ }
+
+ fn convert_to_text(&self, payload: &Payload) -> Result<Payload, Error> {
+ match payload {
+ Payload::Json(value) => {
+ let text = simd_json::to_string_pretty(value)
Review Comment:
Maybe just use `to_string()` for some additional performance reasons?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "spetz (via GitHub)" <gi...@apache.org>.
spetz commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2159480199
##########
core/connectors/sdk/examples/protobuf_config.toml:
##########
@@ -0,0 +1,43 @@
+[iggy]
+address = "localhost:8090"
+username = "iggy"
+password = "iggy"
+
+[sources.protobuf_source]
Review Comment:
Let's add the example `protobuf` config as the separate section under the existing `README.md` file in `connectors/sdk`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on PR #1886:
URL: https://github.com/apache/iggy/pull/1886#issuecomment-3001832391
@hubcio Should be done until Thursday
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(connectors): add protobuf support to connectors runtime [iggy]
Posted by "jaroslawroszyk (via GitHub)" <gi...@apache.org>.
jaroslawroszyk commented on code in PR #1886:
URL: https://github.com/apache/iggy/pull/1886#discussion_r2173282278
##########
core/connectors/sdk/src/transforms/proto_convert.rs:
##########
@@ -0,0 +1,143 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+use prost::Message;
+use prost_types::Any;
+use base64::Engine;
+use serde::{Deserialize, Serialize};
+use simd_json::OwnedValue;
+use tracing::error;
+
+use crate::{DecodedMessage, Error, Payload, TopicMetadata};
+
+use super::{Transform, TransformType};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ProtoConvertConfig {
+ pub target_format: ConvertFormat,
+ pub preserve_structure: bool,
+ pub field_mappings: Option<std::collections::HashMap<String, String>>,
+}
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ConvertFormat {
+ Json,
+ Proto,
+ Text,
+}
+
+pub struct ProtoConvert {
+ config: ProtoConvertConfig,
+}
+
+impl ProtoConvert {
+ pub fn new(config: ProtoConvertConfig) -> Self {
+ Self { config }
+ }
+
+ fn convert_proto_to_json(&self, payload: &[u8]) -> Result<Payload, Error> {
+ match Any::decode(payload) {
+ Ok(any) => {
+ let json_value = simd_json::json!({
+ "type_url": any.type_url,
+ "value": base64::engine::general_purpose::STANDARD.encode(&any.value),
+ "converted_at": chrono::Utc::now().to_rfc3339(),
+ });
+ Ok(Payload::Json(json_value))
+ }
+ Err(e) => {
+ error!("Failed to decode protobuf message: {e}");
+ Err(Error::InvalidProtobufPayload)
+ }
+ }
+ }
+
+ fn convert_json_to_proto(&self, json_value: &OwnedValue) -> Result<Payload, Error> {
+ let json_string = simd_json::to_string(json_value)
+ .map_err(|_| Error::InvalidJsonPayload)?;
+
+ let any = Any {
+ type_url: "type.googleapis.com/google.protobuf.StringValue".to_string(),
+ value: json_string.into_bytes(),
+ };
+
+ let proto_bytes = any.encode_to_vec();
+ Ok(Payload::Raw(proto_bytes))
+ }
+
+ fn convert_to_text(&self, payload: &Payload) -> Result<Payload, Error> {
+ match payload {
+ Payload::Json(value) => {
+ let text = simd_json::to_string_pretty(value)
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org