damondouglas commented on code in PR #31637:
URL: https://github.com/apache/beam/pull/31637#discussion_r1645239077
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java:
##########
@@ -68,4 +68,307 @@ public String getName() {
return name;
}
}
+ /** Represents a Solace destination type. */
+ public enum DestinationType {
+ TOPIC,
+ QUEUE,
+ UNKNOWN
+ }
+
+ /** Represents a Solace message destination (either a Topic or a Queue). */
+ @AutoValue
+ @DefaultSchema(AutoValueSchema.class)
+ public abstract static class Destination {
+ /**
+ * Gets the name of the destination.
+ *
+ * @return The destination name.
+ */
+ @SchemaFieldNumber("0")
+ public abstract String getName();
+
+ /**
+ * Gets the type of the destination (TOPIC, QUEUE or UNKNOWN).
+ *
+ * @return The destination type.
+ */
+ @SchemaFieldNumber("1")
+ public abstract DestinationType getType();
+
+ static Builder builder() {
+ return new AutoValue_Solace_Destination.Builder();
+ }
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setName(String name);
+
+ abstract Builder setType(DestinationType type);
+
+ abstract Destination build();
+ }
+ }
+
+ /** Represents a Solace message record with its associated metadata. */
+ @AutoValue
+ @DefaultSchema(AutoValueSchema.class)
+ public abstract static class Record {
+ /**
+ * Gets the unique identifier of the message, a string for an
application-specific message
+ * identifier.
+ *
+ * <p>Mapped from {@link BytesXMLMessage#getApplicationMessageId()}
+ *
+ * @return The message ID, or null if not available.
+ */
+ @SchemaFieldNumber("0")
+ public abstract @Nullable String getMessageId();
+
+ /**
+ * Gets the payload of the message as a ByteString.
+ *
+ * <p>Mapped from {@link BytesXMLMessage#getBytes()}
+ *
+ * @return The message payload.
+ */
+ @SchemaFieldNumber("1")
+ public abstract ByteString getPayload();
+ /**
+ * Gets the destination (topic or queue) to which the message was sent.
+ *
+ * <p>Mapped from {@link BytesXMLMessage#getDestination()}
+ *
+ * @return The destination, or null if not available.
+ */
+ @SchemaFieldNumber("2")
+ public abstract @Nullable Destination getDestination();
+
+ /**
+ * Gets the message expiration time in milliseconds since the Unix epoch.
+ *
+ * <p>A value of 0 indicates the message does not expire.
+ *
+ * <p>Mapped from {@link BytesXMLMessage#getExpiration()}
+ *
+ * @return The expiration timestamp.
+ */
+ @SchemaFieldNumber("3")
+ public abstract long getExpiration();
+
+ /**
+ * Gets the priority level of the message (0-255, higher is more
important). -1 if not set.
Review Comment:
Where in the code are these constraints enforced? Iimplementing one's own
Builder build method with `autoBuild();` would allow for these constraints,
however when doing this, you'll get an error when deriving a Beam Schema/Row.
In such situations, I've ended up using a Pojo instead of Autovalue, though I
prefer AutoValue when instantiating my own classes. I'll support whatever you
decide but thought it would be helpful to enforce this behavior somewhere and
write a few unit tests to validate.
##########
sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.
+ */
+package org.apache.beam.sdk.io.solace.data;
+
+import java.nio.charset.StandardCharsets;
+import org.apache.beam.sdk.io.solace.data.Solace.Destination;
+import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.ByteString;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SolaceTest {
+
+ Destination destination =
+ Solace.Destination.builder()
+ .setName("some destination")
+ .setType(Solace.DestinationType.TOPIC)
+ .build();
+ String messageId = "some message id";
+ long expiration = 123L;
+ int priority = 7;
+ Boolean redelivered = true;
+ Destination replyTo =
+ Solace.Destination.builder()
+ .setName("some reply-to destination")
+ .setType(Solace.DestinationType.TOPIC)
+ .build();
+ long receiveTimestamp = 123456789L;
+ Long senderTimestamp = 987654321L;
+ Long sequenceNumber = 27L;
+ long timeToLive = 34567890L;
+ String payloadString = "some payload";
+ ByteString payload = ByteString.copyFrom(payloadString,
StandardCharsets.UTF_8);
+ String attachmentString = "some attachment";
+ ByteString attachment = ByteString.copyFrom(attachmentString,
StandardCharsets.UTF_8);
+
+ @Test
+ public void testRecordEquality() {
+ Solace.Record obj1 =
+ Solace.Record.builder()
+ .setDestination(destination)
+ .setMessageId(messageId)
+ .setExpiration(expiration)
+ .setPriority(priority)
+ .setRedelivered(redelivered)
+ .setReplyTo(replyTo)
+ .setReceiveTimestamp(receiveTimestamp)
+ .setSenderTimestamp(senderTimestamp)
+ .setSequenceNumber(sequenceNumber)
+ .setTimeToLive(timeToLive)
+ .setPayload(payload)
+ .setAttachmentBytes(attachment)
+ .build();
+
+ Solace.Record obj2 =
+ Solace.Record.builder()
+ .setDestination(destination)
+ .setMessageId(messageId)
+ .setExpiration(expiration)
+ .setPriority(priority)
+ .setRedelivered(redelivered)
+ .setReplyTo(replyTo)
+ .setReceiveTimestamp(receiveTimestamp)
+ .setSenderTimestamp(senderTimestamp)
+ .setSequenceNumber(sequenceNumber)
+ .setTimeToLive(timeToLive)
+ .setPayload(payload)
+ .setAttachmentBytes(attachment)
+ .build();
+
+ Solace.Record obj3 =
+ Solace.Record.builder()
+ .setDestination(destination)
+ .setMessageId(messageId)
+ .setExpiration(expiration)
+ .setPriority(priority)
+ .setRedelivered(!redelivered)
+ .setReplyTo(replyTo)
+ .setReceiveTimestamp(receiveTimestamp)
+ .setSenderTimestamp(senderTimestamp)
+ .setSequenceNumber(sequenceNumber)
+ .setTimeToLive(timeToLive)
+ .setPayload(payload)
+ .setAttachmentBytes(attachment)
+ .build();
+
+ Assert.assertEquals(obj1, obj2);
+ Assert.assertNotEquals(obj1, obj3);
+ Assert.assertEquals(obj1.hashCode(), obj2.hashCode());
+ Assert.assertEquals(obj1.getDestination(), destination);
+ Assert.assertEquals(obj1.getMessageId(), messageId);
+ Assert.assertEquals(obj1.getExpiration(), expiration);
+ Assert.assertEquals(obj1.getPriority(), priority);
+ Assert.assertEquals(obj1.getRedelivered(), redelivered);
+ Assert.assertEquals(obj1.getReplyTo(), replyTo);
+ Assert.assertEquals(obj1.getReceiveTimestamp(), receiveTimestamp);
+ Assert.assertEquals(obj1.getSenderTimestamp(), senderTimestamp);
+ Assert.assertEquals(obj1.getSequenceNumber(), sequenceNumber);
+ Assert.assertEquals(obj1.getTimeToLive(), timeToLive);
+ Assert.assertEquals(obj1.getPayload().toString(StandardCharsets.UTF_8),
payloadString);
+ Assert.assertEquals(
+ obj1.getAttachmentBytes().toString(StandardCharsets.UTF_8),
attachmentString);
+ }
+
+ @Test
+ public void testRecordNullability() {
Review Comment:
Similar to the comment for testEquality, AutoValue will determine what
fields can be null or not using the Nullable annotation. It may be that this
test isn't needed either.
##########
sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/data/SolaceTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.
+ */
+package org.apache.beam.sdk.io.solace.data;
+
+import java.nio.charset.StandardCharsets;
+import org.apache.beam.sdk.io.solace.data.Solace.Destination;
+import org.apache.beam.vendor.grpc.v1p60p1.com.google.protobuf.ByteString;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SolaceTest {
+
+ Destination destination =
+ Solace.Destination.builder()
+ .setName("some destination")
+ .setType(Solace.DestinationType.TOPIC)
+ .build();
+ String messageId = "some message id";
+ long expiration = 123L;
+ int priority = 7;
+ Boolean redelivered = true;
+ Destination replyTo =
+ Solace.Destination.builder()
+ .setName("some reply-to destination")
+ .setType(Solace.DestinationType.TOPIC)
+ .build();
+ long receiveTimestamp = 123456789L;
+ Long senderTimestamp = 987654321L;
+ Long sequenceNumber = 27L;
+ long timeToLive = 34567890L;
+ String payloadString = "some payload";
+ ByteString payload = ByteString.copyFrom(payloadString,
StandardCharsets.UTF_8);
+ String attachmentString = "some attachment";
+ ByteString attachment = ByteString.copyFrom(attachmentString,
StandardCharsets.UTF_8);
+
+ @Test
+ public void testRecordEquality() {
Review Comment:
If you are using AutoValue, it automatically creates an equals method
possibly removing the need for this test.
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java:
##########
@@ -68,4 +68,307 @@ public String getName() {
return name;
}
}
+ /** Represents a Solace destination type. */
+ public enum DestinationType {
+ TOPIC,
+ QUEUE,
+ UNKNOWN
+ }
+
+ /** Represents a Solace message destination (either a Topic or a Queue). */
+ @AutoValue
+ @DefaultSchema(AutoValueSchema.class)
+ public abstract static class Destination {
+ /**
+ * Gets the name of the destination.
+ *
+ * @return The destination name.
+ */
+ @SchemaFieldNumber("0")
+ public abstract String getName();
+
+ /**
+ * Gets the type of the destination (TOPIC, QUEUE or UNKNOWN).
+ *
+ * @return The destination type.
+ */
+ @SchemaFieldNumber("1")
+ public abstract DestinationType getType();
+
+ static Builder builder() {
+ return new AutoValue_Solace_Destination.Builder();
+ }
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setName(String name);
+
+ abstract Builder setType(DestinationType type);
+
+ abstract Destination build();
+ }
+ }
+
+ /** Represents a Solace message record with its associated metadata. */
+ @AutoValue
+ @DefaultSchema(AutoValueSchema.class)
+ public abstract static class Record {
+ /**
+ * Gets the unique identifier of the message, a string for an
application-specific message
+ * identifier.
+ *
+ * <p>Mapped from {@link BytesXMLMessage#getApplicationMessageId()}
+ *
+ * @return The message ID, or null if not available.
+ */
+ @SchemaFieldNumber("0")
+ public abstract @Nullable String getMessageId();
Review Comment:
What would Solace respond with if the message ID was an empty string instead
of null?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
