|
18 | 18 | package org.apache.beam.sdk.extensions.gcp.options; |
19 | 19 |
|
20 | 20 | import com.fasterxml.jackson.annotation.JsonIgnore; |
| 21 | +import com.fasterxml.jackson.core.JsonGenerator; |
| 22 | +import com.fasterxml.jackson.core.JsonParser; |
| 23 | +import com.fasterxml.jackson.databind.DeserializationContext; |
| 24 | +import com.fasterxml.jackson.databind.JsonDeserializer; |
| 25 | +import com.fasterxml.jackson.databind.JsonNode; |
| 26 | +import com.fasterxml.jackson.databind.JsonSerializer; |
| 27 | +import com.fasterxml.jackson.databind.ObjectMapper; |
| 28 | +import com.fasterxml.jackson.databind.SerializerProvider; |
| 29 | +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; |
| 30 | +import com.fasterxml.jackson.databind.annotation.JsonSerialize; |
21 | 31 | import com.google.cloud.hadoop.gcsio.GoogleCloudStorageReadOptions; |
22 | 32 | import com.google.cloud.hadoop.util.AsyncWriteChannelOptions; |
| 33 | +import java.lang.reflect.Method; |
23 | 34 | import java.util.HashMap; |
24 | 35 | import java.util.concurrent.ExecutorService; |
25 | 36 | import org.apache.beam.sdk.extensions.gcp.storage.GcsPathValidator; |
@@ -53,8 +64,48 @@ public GoogleCloudStorageReadOptions create(PipelineOptions options) { |
53 | 64 | } |
54 | 65 | } |
55 | 66 |
|
| 67 | + class GcsReadOptionsSerializer extends JsonSerializer<GoogleCloudStorageReadOptions> { |
| 68 | + @Override |
| 69 | + public void serialize( |
| 70 | + GoogleCloudStorageReadOptions value, JsonGenerator gen, SerializerProvider serializers) |
| 71 | + throws java.io.IOException { |
| 72 | + serializers.defaultSerializeValue(value, gen); |
| 73 | + } |
| 74 | + } |
| 75 | + |
| 76 | + class GcsReadOptionsDeserializer extends JsonDeserializer<GoogleCloudStorageReadOptions> { |
| 77 | + @Override |
| 78 | + public GoogleCloudStorageReadOptions deserialize(JsonParser p, DeserializationContext ctxt) |
| 79 | + throws java.io.IOException { |
| 80 | + ObjectMapper mapper = (ObjectMapper) p.getCodec(); |
| 81 | + JsonNode root = mapper.readTree(p); |
| 82 | + GoogleCloudStorageReadOptions.Builder builder = GoogleCloudStorageReadOptions.builder(); |
| 83 | + |
| 84 | + for (Method method : GoogleCloudStorageReadOptions.Builder.class.getMethods()) { |
| 85 | + if (method.getName().startsWith("set") && method.getParameterCount() == 1) { |
| 86 | + String propName = |
| 87 | + Character.toLowerCase(method.getName().charAt(3)) + method.getName().substring(4); |
| 88 | + JsonNode node = root.get(propName); |
| 89 | + if (node != null && !node.isNull()) { |
| 90 | + try { |
| 91 | + Class<?> paramType = method.getParameterTypes()[0]; |
| 92 | + Object val = mapper.treeToValue(node, paramType); |
| 93 | + if (val != null) { |
| 94 | + method.invoke(builder, java.util.Objects.requireNonNull(val)); |
| 95 | + } |
| 96 | + } catch (Exception ignored) { |
| 97 | + // Ignore incompatible or unmappable properties |
| 98 | + } |
| 99 | + } |
| 100 | + } |
| 101 | + } |
| 102 | + return builder.build(); |
| 103 | + } |
| 104 | + } |
| 105 | + |
56 | 106 | /** @deprecated This option will be removed in a future release. */ |
57 | | - @JsonIgnore |
| 107 | + @JsonSerialize(using = GcsReadOptionsSerializer.class) |
| 108 | + @JsonDeserialize(using = GcsReadOptionsDeserializer.class) |
58 | 109 | @Description( |
59 | 110 | "The GoogleCloudStorageReadOptions instance that should be used to read from Google Cloud Storage.") |
60 | 111 | @Default.InstanceFactory(GcsReadOptionsFactory.class) |
|
0 commit comments