Skip to content

Commit 0b9fcd3

Browse files
stan-r-dsstan-r-ds
committed
Improve handling of ByteBuffer fields
1 parent dd8a24b commit 0b9fcd3

39 files changed

Lines changed: 315 additions & 185 deletions

File tree

‎core/src/main/java/dev/streamx/cli/command/ingestion/BaseIngestionCommand.java‎

Lines changed: 0 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,18 @@
11
package dev.streamx.cli.command.ingestion;
22

3-
import static java.nio.charset.StandardCharsets.UTF_8;
4-
5-
import com.fasterxml.jackson.databind.JsonNode;
6-
import com.fasterxml.jackson.databind.ObjectMapper;
7-
import com.fasterxml.jackson.databind.node.ObjectNode;
83
import com.streamx.clients.ingestion.StreamxClient;
94
import com.streamx.clients.ingestion.exceptions.StreamxClientException;
105
import com.streamx.clients.ingestion.publisher.Publisher;
116
import dev.streamx.cli.exception.IngestionClientException;
127
import dev.streamx.cli.util.ExceptionUtils;
13-
import io.cloudevents.CloudEvent;
14-
import io.cloudevents.jackson.JsonCloudEventData;
158
import jakarta.inject.Inject;
16-
import java.util.Base64;
179
import javax.net.ssl.SSLHandshakeException;
1810
import picocli.CommandLine.ArgGroup;
1911
import picocli.CommandLine.Model.CommandSpec;
2012
import picocli.CommandLine.Spec;
2113

2214
public abstract class BaseIngestionCommand implements Runnable {
2315

24-
protected static final ObjectMapper objectMapper = new ObjectMapper();
25-
2616
@ArgGroup(exclusive = false)
2717
IngestionArguments ingestionArguments;
2818

@@ -49,31 +39,4 @@ public final void run() {
4939
throw ExceptionUtils.sneakyThrow(e);
5040
}
5141
}
52-
53-
protected static CloudEvent withAdjustedData(CloudEvent event) {
54-
JsonCloudEventData data = (JsonCloudEventData) event.getData();
55-
if (data != null) {
56-
JsonNode dataNode = data.getNode();
57-
if (dataNode != null) {
58-
JsonNode contentNode = dataNode.get("content");
59-
if (contentNode != null) {
60-
JsonNode bytesNode = contentNode.get("bytes");
61-
if (bytesNode != null) {
62-
String content = bytesNode.asText();
63-
if (content != null) {
64-
// TODO: detect if the content.bytes field is already base 64 encoded
65-
// and if it is encoded - decode it first.
66-
// For now assume it's in clear text
67-
String encoded = Base64.getEncoder().encodeToString(content.getBytes(UTF_8));
68-
69-
ObjectNode dataNodeObj = (ObjectNode) dataNode;
70-
dataNodeObj.put("content", encoded);
71-
return CloudEventBuilder.copyWithNewData(event, dataNodeObj);
72-
}
73-
}
74-
}
75-
}
76-
}
77-
return event;
78-
}
7942
}

‎core/src/main/java/dev/streamx/cli/command/ingestion/CloudEventBuilder.java‎

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -41,11 +41,4 @@ public static CloudEvent build(String subject, String type, String source, JsonN
4141
.build();
4242
}
4343

44-
public static CloudEvent copyWithNewData(CloudEvent cloudEvent, JsonNode data) {
45-
return io.cloudevents.core.builder.CloudEventBuilder.v1(cloudEvent)
46-
.withDataContentType("application/json")
47-
.withData(JsonCloudEventData.wrap(data))
48-
.build();
49-
}
50-
5144
}

‎core/src/main/java/dev/streamx/cli/command/ingestion/batch/BatchCommand.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -100,10 +100,9 @@ private void send(State state, Publisher publisher) throws StreamxClientExceptio
100100
state.eventSubject,
101101
state.eventType,
102102
state.eventSource,
103-
state.payload
103+
state.payload // TODO rename payload to data everywhere
104104
);
105-
CloudEvent adjustedEvent = withAdjustedData(inputEvent);
106-
CloudEvent responseEvent = publisher.send(adjustedEvent);
105+
CloudEvent responseEvent = publisher.send(inputEvent);
107106

108107
printf("Sent %s event using batch with key '%s' at %s%n",
109108
state.eventType, state.eventSubject(),

‎core/src/main/java/dev/streamx/cli/command/ingestion/batch/resolver/step/BinaryResolverStep.java‎

Lines changed: 21 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,23 @@
11
package dev.streamx.cli.command.ingestion.batch.resolver.step;
22

33
import com.fasterxml.jackson.databind.JsonNode;
4-
import com.fasterxml.jackson.databind.ObjectMapper;
54
import com.fasterxml.jackson.databind.node.ArrayNode;
5+
import com.fasterxml.jackson.databind.node.BinaryNode;
66
import com.fasterxml.jackson.databind.node.ObjectNode;
7-
import com.fasterxml.jackson.databind.util.TokenBuffer;
87
import dev.streamx.cli.exception.PayloadException;
9-
import dev.streamx.cli.util.FileSourceUtils;
108
import jakarta.enterprise.context.ApplicationScoped;
119
import java.io.IOException;
12-
import java.nio.charset.StandardCharsets;
10+
import java.nio.file.Files;
11+
import java.nio.file.NoSuchFileException;
12+
import java.nio.file.Path;
1313
import java.util.Iterator;
1414
import java.util.Map;
1515
import java.util.Map.Entry;
1616

1717
@ApplicationScoped
1818
public class BinaryResolverStep implements ResolverStep {
1919

20-
private static final ObjectMapper BINARY_SERIALIZATION_OBJECT_MAPPER = new ObjectMapper();
20+
private static final String FILE_STRATEGY_PREFIX = "file://";
2121

2222
@Override
2323
public JsonNode resolve(JsonNode payload, Map<String, String> variables) {
@@ -31,8 +31,9 @@ public JsonNode resolve(JsonNode payload, Map<String, String> variables) {
3131
private JsonNode evaluateBinary(JsonNode evaluatedPayload) {
3232
if (evaluatedPayload.isTextual()) {
3333
String text = evaluatedPayload.textValue();
34-
if (FileSourceUtils.applies(text)) {
35-
return toJsonNode(FileSourceUtils.resolve(text));
34+
if (isFileReference(text)) {
35+
byte[] content = readFileContent(text);
36+
return new BinaryNode(content);
3637
} else {
3738
return evaluatedPayload;
3839
}
@@ -55,18 +56,20 @@ private JsonNode evaluateBinary(JsonNode evaluatedPayload) {
5556
}
5657
}
5758

58-
/**
59-
* Converts binary data into a JSON node by writing it as an ISO_8859_1 string.
60-
*/
61-
private static JsonNode toJsonNode(byte[] datum) {
62-
if (datum == null) {
63-
return null;
64-
}
65-
try (var generator = new TokenBuffer(BINARY_SERIALIZATION_OBJECT_MAPPER, false)) {
66-
generator.writeString(new String(datum, StandardCharsets.ISO_8859_1));
67-
return BINARY_SERIALIZATION_OBJECT_MAPPER.readTree(generator.asParser());
59+
public static boolean isFileReference(String rawSource) {
60+
return rawSource != null && rawSource.startsWith(FILE_STRATEGY_PREFIX);
61+
}
62+
63+
public static byte[] readFileContent(String rawSource) {
64+
String sourceFile = rawSource.substring(FILE_STRATEGY_PREFIX.length());
65+
Path path = Path.of(sourceFile);
66+
try {
67+
return Files.readAllBytes(path);
68+
} catch (NoSuchFileException e) {
69+
throw PayloadException.noSuchFileException(e, path);
6870
} catch (IOException e) {
69-
throw PayloadException.ioException(e);
71+
throw PayloadException.fileReadingException(e, path);
7072
}
7173
}
74+
7275
}

‎core/src/main/java/dev/streamx/cli/command/ingestion/stream/StreamCommand.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,12 +4,14 @@
44

55
import com.fasterxml.jackson.core.JsonProcessingException;
66
import com.fasterxml.jackson.databind.JsonNode;
7+
import com.fasterxml.jackson.databind.ObjectMapper;
78
import com.streamx.ce.serialization.DeserializerException;
89
import com.streamx.ce.serialization.json.CloudEventJsonDeserializer;
910
import com.streamx.clients.ingestion.exceptions.StreamxClientException;
1011
import com.streamx.clients.ingestion.publisher.Publisher;
1112
import dev.streamx.cli.VersionProvider;
1213
import dev.streamx.cli.command.ingestion.BaseIngestionCommand;
14+
import dev.streamx.cli.command.ingestion.stream.parser.JsonBase64Encoder;
1315
import dev.streamx.cli.command.ingestion.stream.parser.StreamIngestionJsonParser;
1416
import dev.streamx.cli.util.ExceptionUtils;
1517
import dev.streamx.cli.util.FileUtils;
@@ -19,6 +21,7 @@
1921
import java.io.IOException;
2022
import java.nio.file.Path;
2123
import java.nio.file.Paths;
24+
import java.util.List;
2225
import picocli.CommandLine.ArgGroup;
2326
import picocli.CommandLine.Command;
2427

@@ -32,6 +35,7 @@ public class StreamCommand extends BaseIngestionCommand {
3235
// TODO "_v2" is a temporary postfix for now
3336
public static final String COMMAND_NAME = "stream_v2";
3437

38+
private static final ObjectMapper objectMapper = new ObjectMapper();
3539
private static final CloudEventJsonDeserializer deserializer = new CloudEventJsonDeserializer();
3640

3741
@ArgGroup(exclusive = false, multiplicity = "1")
@@ -43,10 +47,13 @@ public class StreamCommand extends BaseIngestionCommand {
4347
@Override
4448
protected void perform(Publisher publisher) throws StreamxClientException {
4549
Path streamFile = Paths.get(streamIngestionArguments.getSourceFile());
50+
List<String> jsonPathsToEncodeToBase64 = StreamProperties
51+
.getJsonPathsToEncodeToBase64(streamFile);
4652

4753
try (FileInputStream fis = new FileInputStream(streamFile.toFile())) {
4854

4955
ingestionJsonParser.parse(fis, cloudEventNode -> {
56+
JsonBase64Encoder.encodeFields(cloudEventNode, jsonPathsToEncodeToBase64);
5057
CloudEvent inputEvent = toCloudEvent(cloudEventNode);
5158
CloudEvent responseEvent = publisher.send(inputEvent);
5259
printf("Sent %s event using stream with key '%s' at %s%n",
@@ -66,7 +73,6 @@ protected void perform(Publisher publisher) throws StreamxClientException {
6673

6774
private static CloudEvent toCloudEvent(JsonNode cloudEventNode) throws JsonProcessingException {
6875
byte[] cloudEventJsonBytes = objectMapper.writeValueAsBytes(cloudEventNode);
69-
CloudEvent inputEvent = deserializer.deserialize(cloudEventJsonBytes);
70-
return withAdjustedData(inputEvent);
76+
return deserializer.deserialize(cloudEventJsonBytes);
7177
}
7278
}

‎core/src/main/java/dev/streamx/cli/command/ingestion/stream/StreamProperties.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
package dev.streamx.cli.command.ingestion.stream;
2+
3+
import java.io.IOException;
4+
import java.io.InputStream;
5+
import java.nio.file.Files;
6+
import java.nio.file.Path;
7+
import java.util.Arrays;
8+
import java.util.Collections;
9+
import java.util.List;
10+
import java.util.Properties;
11+
import org.apache.commons.lang3.StringUtils;
12+
import org.jboss.logging.Logger;
13+
14+
class StreamProperties {
15+
16+
private static final Logger logger = Logger.getLogger(StreamProperties.class);
17+
private static final String PROPERTIES_FILE_NAME = ".stream.properties";
18+
private static final String BASE64_PATHS_PROPERTY_NAME = "json.paths.to.encode.to.base64";
19+
20+
private StreamProperties() {
21+
// no instances
22+
}
23+
24+
public static List<String> getJsonPathsToEncodeToBase64(Path streamFile) {
25+
Path propertiesFile = streamFile.resolveSibling(PROPERTIES_FILE_NAME);
26+
if (Files.exists(propertiesFile)) {
27+
try (InputStream fileStream = Files.newInputStream(propertiesFile)) {
28+
Properties properties = new Properties();
29+
properties.load(fileStream);
30+
String paths = properties.getProperty(BASE64_PATHS_PROPERTY_NAME);
31+
if (StringUtils.isNotBlank(paths)) {
32+
return Arrays.stream(paths.split(","))
33+
.map(String::trim)
34+
.filter(StringUtils::isNotEmpty)
35+
.toList();
36+
}
37+
} catch (IOException ex) {
38+
logger.warn("Error reading properties from " + propertiesFile, ex);
39+
}
40+
}
41+
return Collections.emptyList();
42+
}
43+
44+
}

‎core/src/main/java/dev/streamx/cli/command/ingestion/stream/parser/JsonBase64Encoder.java‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
package dev.streamx.cli.command.ingestion.stream.parser;
2+
3+
import com.fasterxml.jackson.databind.JsonNode;
4+
import com.fasterxml.jackson.databind.node.ObjectNode;
5+
import com.fasterxml.jackson.databind.node.TextNode;
6+
import java.util.Base64;
7+
import java.util.List;
8+
9+
public class JsonBase64Encoder {
10+
11+
private JsonBase64Encoder() {
12+
// no instances
13+
}
14+
15+
public static void encodeFields(JsonNode root, List<String> jsonPaths) {
16+
for (String jsonPath : jsonPaths) {
17+
JsonNode node = root.at(jsonPath);
18+
19+
if (node.isMissingNode() || !node.isTextual()) {
20+
continue;
21+
}
22+
23+
TextNode encodedTextNode = createTextNodeWithEncodedContent(node);
24+
replaceNode(root, jsonPath, encodedTextNode);
25+
}
26+
}
27+
28+
private static TextNode createTextNodeWithEncodedContent(JsonNode node) {
29+
String encodedText = Base64.getEncoder().encodeToString(node.asText().getBytes());
30+
return TextNode.valueOf(encodedText);
31+
}
32+
33+
private static void replaceNode(JsonNode root, String jsonPath, JsonNode newValue) {
34+
int lastSlash = jsonPath.lastIndexOf('/');
35+
String parentNodePath = jsonPath.substring(0, lastSlash);
36+
String fieldName = jsonPath.substring(lastSlash + 1);
37+
38+
JsonNode parentNode = root.at(parentNodePath);
39+
if (parentNode instanceof ObjectNode obj) {
40+
obj.set(fieldName, newValue);
41+
}
42+
}
43+
}

‎core/src/main/java/dev/streamx/cli/exception/PayloadException.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -17,15 +17,6 @@ private PayloadException(String message) {
1717
super(message);
1818
}
1919

20-
public static PayloadException payloadNotFound() {
21-
return new PayloadException("""
22-
Payload definition not found.
23-
24-
Make sure that:
25-
* there is an optional third parameter that points to the file containing the payload,
26-
* or there is at least one Payload Defining Option (-s/-b/-j)""");
27-
}
28-
2920
public static PayloadException jsonParseException(JsonParseException exception, String payload) {
3021
return new PayloadException("""
3122
Payload could not be parsed.

‎core/src/main/java/dev/streamx/cli/util/FileSourceUtils.java‎

Lines changed: 0 additions & 38 deletions
This file was deleted.

0 commit comments

Comments
 (0)