[proxima-scheme-proto] #202 add proto scheme provider for transactions

scheme/pom.xml

@@ -64,15 +64,20 @@
       <artifactId>slf4j-api</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-core</artifactId>
+      <version>${project.version}</version>
+      <classifier>tests</classifier>
+      <scope>test</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-log4j12</artifactId>
       <scope>test</scope>
     </dependency>
 
-
-
-
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

scheme/proto/src/main/java/cz/o2/proxima/scheme/proto/ProtoSerializerFactory.java

@@ -15,16 +15,25 @@
  */
 package cz.o2.proxima.scheme.proto;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.AbstractMessage;
 import com.google.protobuf.Message.Builder;
 import com.google.protobuf.Parser;
 import com.google.protobuf.TextFormat;
 import com.google.protobuf.util.JsonFormat;
+import cz.o2.proxima.functional.UnaryFunction;
 import cz.o2.proxima.scheme.AttributeValueAccessor;
 import cz.o2.proxima.scheme.SchemaDescriptors.SchemaTypeDescriptor;
 import cz.o2.proxima.scheme.ValueSerializer;
 import cz.o2.proxima.scheme.ValueSerializerFactory;
+import cz.o2.proxima.scheme.proto.transactions.Transactions.ProtoRequest;
+import cz.o2.proxima.scheme.proto.transactions.Transactions.ProtoResponse;
+import cz.o2.proxima.scheme.proto.transactions.Transactions.ProtoState;
 import cz.o2.proxima.scheme.proto.utils.ProtoUtils;
+import cz.o2.proxima.transaction.Request;
+import cz.o2.proxima.transaction.Response;
+import cz.o2.proxima.transaction.State;
+import cz.o2.proxima.transaction.TransactionSerializerSchemeProvider;
 import cz.o2.proxima.util.Classpath;
 import cz.o2.proxima.util.ExceptionUtils;
 import java.lang.reflect.InvocationTargetException;
@@ -51,7 +60,11 @@ public String getAcceptableScheme() {
 
   @SuppressWarnings("unchecked")
   private static <M extends AbstractMessage> ValueSerializer<M> createSerializer(URI uri) {
-    return new ProtoValueSerializer<>(uri.getSchemeSpecificPart());
+    String className = uri.getSchemeSpecificPart();
+    if (className.startsWith("cz.o2.proxima.transaction.")) {
+      return TransactionProtoSerializer.ofTransactionClass(className);
+    }
+    return new ProtoValueSerializer<>(className);
   }
 
   @SuppressWarnings("unchecked")
@@ -73,6 +86,19 @@ public <T> ValueSerializer<T> getValueSerializer(URI scheme) {
         parsers.computeIfAbsent(scheme, ProtoSerializerFactory::createSerializer);
   }
 
+  @Override
+  public boolean canProvideTransactionSerializer() {
+    return true;
+  }
+
+  @Override
+  public TransactionSerializerSchemeProvider createTransactionSerializerSchemeProvider() {
+    return TransactionSerializerSchemeProvider.of(
+        "proto:" + Request.class.getName(),
+        "proto:" + Response.class.getName(),
+        "proto:" + State.class.getName());
+  }
+
   private static class ProtoValueSerializer<MessageT extends AbstractMessage>
       implements ValueSerializer<MessageT> {
 
@@ -171,4 +197,62 @@ public <OutputT> AttributeValueAccessor<MessageT, OutputT> getValueAccessor() {
       return (AttributeValueAccessor<MessageT, OutputT>) accessor;
     }
   }
+
+  @VisibleForTesting
+  static class TransactionProtoSerializer<T> implements ValueSerializer<T> {
+
+    @SuppressWarnings("unchecked")
+    static <V> TransactionProtoSerializer<V> ofTransactionClass(String className) {
+      switch (className) {
+        case "cz.o2.proxima.transaction.Request":
+          return (TransactionProtoSerializer<V>)
+              new TransactionProtoSerializer<>(
+                  new ProtoValueSerializer<>(ProtoRequest.class.getName()),
+                  req -> ProtoRequest.newBuilder().build(),
+                  req -> Request.of());
+        case "cz.o2.proxima.transaction.Response":
+          return (TransactionProtoSerializer<V>)
+              new TransactionProtoSerializer<>(
+                  new ProtoValueSerializer<>(ProtoResponse.class.getName()),
+                  req -> ProtoResponse.newBuilder().build(),
+                  req -> Response.of());
+        case "cz.o2.proxima.transaction.State":
+          return (TransactionProtoSerializer<V>)
+              new TransactionProtoSerializer<>(
+                  new ProtoValueSerializer<>(ProtoState.class.getName()),
+                  req -> ProtoState.newBuilder().build(),
+                  req -> State.of());
+      }
+      throw new UnsupportedOperationException("Unknown className of transactions: " + className);
+    }
+
+    private final ProtoValueSerializer<AbstractMessage> inner;
+    private final UnaryFunction<T, AbstractMessage> asMessage;
+    private final UnaryFunction<AbstractMessage, T> asTransaction;
+
+    public TransactionProtoSerializer(
+        ProtoValueSerializer<AbstractMessage> inner,
+        UnaryFunction<T, AbstractMessage> asMessage,
+        UnaryFunction<AbstractMessage, T> asTransaction) {
+
+      this.inner = inner;
+      this.asMessage = asMessage;
+      this.asTransaction = asTransaction;
+    }
+
+    @Override
+    public Optional<T> deserialize(byte[] input) {
+      return inner.deserialize(input).map(asTransaction::apply);
+    }
+
+    @Override
+    public byte[] serialize(T value) {
+      return inner.serialize(asMessage.apply(value));
+    }
+
+    @Override
+    public T getDefault() {
+      return asTransaction.apply(inner.getDefault());
+    }
+  }
 }

scheme/proto/src/main/proto/transactions.proto

@@ -0,0 +1,30 @@
+/**
+ * Copyright 2017-2021 O2 Czech Republic, a.s.
+ *
+ * Licensed 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.
+ */
+syntax = "proto3";
+
+package cz.o2.proxima.scheme.proto.transactions;
+
+message ProtoRequest {
+
+}
+
+message ProtoResponse {
+
+}
+
+message ProtoState {
+
+}

scheme/proto/src/test/java/cz/o2/proxima/scheme/proto/ProtoSerializerFactoryTest.java

@@ -15,17 +15,24 @@
  */
 package cz.o2.proxima.scheme.proto;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 import com.google.protobuf.ByteString;
+import com.typesafe.config.ConfigFactory;
+import cz.o2.proxima.repository.AttributeDescriptor;
+import cz.o2.proxima.repository.EntityDescriptor;
+import cz.o2.proxima.repository.Repository;
 import cz.o2.proxima.scheme.AttributeValueAccessor;
 import cz.o2.proxima.scheme.AttributeValueAccessors.StructureValue;
 import cz.o2.proxima.scheme.AttributeValueType;
 import cz.o2.proxima.scheme.SchemaDescriptors.SchemaTypeDescriptor;
 import cz.o2.proxima.scheme.ValueSerializer;
 import cz.o2.proxima.scheme.ValueSerializerFactory;
+import cz.o2.proxima.scheme.proto.ProtoSerializerFactory.TransactionProtoSerializer;
 import cz.o2.proxima.scheme.proto.test.Scheme.Event;
+import cz.o2.proxima.transaction.Request;
+import cz.o2.proxima.transaction.Response;
+import cz.o2.proxima.transaction.State;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
@@ -104,4 +111,34 @@ public void testGetValueAccessor() {
     assertEquals("gatewayId value", created.getGatewayId());
     assertEquals("payload value", created.getPayload().toStringUtf8());
   }
+
+  @Test
+  public void testTransactionSchemeProvider() {
+    Repository repo =
+        Repository.ofTest(
+            ConfigFactory.load("test-transactions-proto.conf")
+                .withFallback(ConfigFactory.load("test-transactions.conf"))
+                .resolve());
+    EntityDescriptor transaction = repo.getEntity("_transaction");
+    AttributeDescriptor<Request> request = transaction.getAttribute("request.*");
+    assertTrue(request.getValueSerializer() instanceof TransactionProtoSerializer);
+    assertTrue(request.getValueSerializer().isUsable());
+    byte[] bytes = request.getValueSerializer().serialize(Request.of());
+    assertNotNull(bytes);
+    assertTrue(request.getValueSerializer().deserialize(bytes).isPresent());
+
+    AttributeDescriptor<Response> response = transaction.getAttribute("response.*");
+    assertTrue(response.getValueSerializer() instanceof TransactionProtoSerializer);
+    assertTrue(request.getValueSerializer().isUsable());
+    bytes = response.getValueSerializer().serialize(Response.of());
+    assertNotNull(bytes);
+    assertTrue(response.getValueSerializer().deserialize(bytes).isPresent());
+
+    AttributeDescriptor<State> state = transaction.getAttribute("state");
+    assertTrue(state.getValueSerializer() instanceof TransactionProtoSerializer);
+    assertTrue(state.getValueSerializer().isUsable());
+    bytes = state.getValueSerializer().serialize(State.of());
+    assertNotNull(bytes);
+    assertTrue(state.getValueSerializer().deserialize(bytes).isPresent());
+  }
 }

scheme/proto/src/test/resources/test-transactions-proto.conf

@@ -0,0 +1,3 @@
+transactions {
+  scheme-provider: proto
+}