stankiewicz commented on code in PR #39876:
URL: https://github.com/apache/beam/pull/39876#discussion_r3879941931
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/data/Solace.java:
##########
@@ -462,5 +497,118 @@ public static class SolaceRecordMapper {
}
return destinationBuilder.build();
}
+
+ /**
+ * Maps a {@link Record} to a {@link BytesXMLMessage}.
+ *
+ * <p>Only the fields common to both a {@link Record} and a {@link
BytesXMLMessage} are set: the
+ * payload (according to the record's {@link Record.PayloadType}), the
sender timestamp
+ * (defaulting to the current time when the record does not provide one)
and the application
+ * message id. Publishing-specific fields such as delivery mode or
correlation key are not
+ * handled here and must be set by the caller.
+ *
+ * @param record the {@link Record} to map.
+ * @return a JCSMP {@link BytesXMLMessage} carrying the record's common
fields.
+ */
+ public static BytesXMLMessage toMessage(Record record) {
+ BytesXMLMessage msg = encodePayload(record);
+
+ Long senderTimestamp = record.getSenderTimestamp();
+ if (senderTimestamp == null) {
+ senderTimestamp = System.currentTimeMillis();
+ }
+ msg.setSenderTimestamp(senderTimestamp);
+ msg.setApplicationMessageId(record.getMessageId());
+
+ return msg;
+ }
+
+ /**
+ * Reads the payload from a {@link Solace.Record} into a
partially-populated {@link
+ * BytesXMLMessage}.
+ *
+ * @param record the Solace record.
+ * @return a {@link BytesXMLMessage} with the payload set based on the
record's payload type.
+ */
+ private static BytesXMLMessage encodePayload(Record record) {
+ switch (record.getPayloadType()) {
+ case TEXT:
+ TextMessage text =
JCSMPFactory.onlyInstance().createMessage(TextMessage.class);
+ text.setText(record.getText());
+ return text;
+ case BYTES:
+ BytesMessage bytes =
JCSMPFactory.onlyInstance().createMessage(BytesMessage.class);
+ bytes.setData(record.getPayload());
+ return bytes;
+ case BYTES_XML:
+ BytesXMLMessage xml =
JCSMPFactory.onlyInstance().createBytesXMLMessage();
+ xml.writeBytes(record.getPayload());
+ if (record.getAttachmentBytes().length != 0) {
+ xml.writeAttachment(record.getAttachmentBytes());
+ }
+ return xml;
+ default:
+ throw new IllegalArgumentException(
+ "Unsupported payload type: " + record.getPayloadType());
+ }
+ }
+
+ /**
+ * Reads the payload from a {@link BytesXMLMessage} into a
partially-populated {@link
+ * Record.Builder}.
+ *
+ * @param msg the JCSMP message.
+ * @return a {@link Record.Builder} with the payload and payload type set
based on the message
+ * type.
+ */
+ private static Record.Builder decodePayload(@NonNull BytesXMLMessage msg) {
+ if (msg instanceof TextMessage) {
+ String text = ((TextMessage) msg).getText();
+ byte[] payload = text == null ? new byte[0] :
text.getBytes(StandardCharsets.UTF_8);
+ return
Record.builder().setPayloadType(Record.PayloadType.TEXT).setPayload(payload);
+ }
+
+ if (msg instanceof BytesMessage) {
+ byte[] data = ((BytesMessage) msg).getData();
+ byte[] payload = data == null ? new byte[0] : data;
+ return
Record.builder().setPayloadType(Record.PayloadType.BYTES).setPayload(payload);
+ }
+
+ // BYTES_XML fallback
+ byte[] payload = readBytes(msg);
+ byte[] attachment = readAttachment(msg);
+ return Record.builder()
+ .setPayloadType(Record.PayloadType.BYTES_XML)
+ .setPayload(payload)
+ .setAttachmentBytes(attachment);
+ }
+
+ private static byte[] readBytes(BytesXMLMessage msg) {
+ if (msg.getContentLength() == 0) {
+ return new byte[0];
+ }
+ try {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ out.write(msg.getBytes());
+ return out.toByteArray();
Review Comment:
@iht let me know if this is safe to optimize. msg.getBytes sometimes
produces new array but sometimes give access to array backing msg.
--
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]
