Adding configuration, updating interfaces.

Author: mindaugasbarcauskas

src/main/java/com/uber/cadence/largeblob/Configuration.java

@@ -0,0 +1,78 @@
+package com.uber.cadence.largeblob;
+
+import com.uber.cadence.converter.DataConverter;
+import com.uber.cadence.converter.JsonDataConverter;
+
+import java.time.Duration;
+
+public class Configuration {
+
+    private Storage storage;
+    private DataConverter dataConverter;
+    private Long maxBytes;
+    private Duration ttl;
+
+    private Configuration() {
+    }
+
+    public static Builder newBuilder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+
+        private Storage storage;
+        private DataConverter dataConverter = JsonDataConverter.getInstance();
+        private Duration ttl;
+        private Long maxBytes = 4096L;
+
+        public Configuration build() {
+            if (storage == null) {
+                throw new IllegalArgumentException("storage must be provided");
+            }
+
+            Configuration configuration = new Configuration();
+            configuration.storage = this.storage;
+            configuration.dataConverter = this.dataConverter;
+            configuration.ttl = this.ttl;
+            configuration.maxBytes = this.maxBytes;
+            return configuration;
+        }
+
+        public Builder setDataConverter(DataConverter dataConverter) {
+            this.dataConverter = dataConverter;
+            return this;
+        }
+
+        public Builder setStorage(Storage storage) {
+            this.storage = storage;
+            return this;
+        }
+
+        public Builder setTtl(Duration ttl) {
+            this.ttl = ttl;
+            return this;
+        }
+
+        public Builder setMaxBytes(Long maxBytes) {
+            this.maxBytes = maxBytes;
+            return this;
+        }
+    }
+
+    public Storage getStorage() {
+        return storage;
+    }
+
+    public DataConverter getDataConverter() {
+        return dataConverter;
+    }
+
+    public Duration getTtl() {
+        return ttl;
+    }
+
+    public Long getMaxBytes() {
+        return maxBytes;
+    }
+}

src/main/java/com/uber/cadence/largeblob/Future.java

@@ -15,64 +15,65 @@
 
 package com.uber.cadence.largeblob;
 
+import com.uber.cadence.converter.DataConverterException;
+
 import java.io.IOException;
 
 /**
  * Future is used for passing around potentially large parameters between activities. Future can never be used in Workflow code because it uses an external storage and that will make
  * workflow code non deterministic. Small amounts of data are stored in the future instance itself, while larger amounts of data are stored in an external storage.
  */
-public class Future {
+public class Future<T> {
 
   private byte[] encoded;
   private String url;
-  private final Storage storage;
-  private final long maxBytesInMemory;
+  private final Configuration config;
+  private final Class<T> clazz;
 
-  public Future(Storage storage, long maxBytesInMemory, byte[] encoded) {
-    this.storage = storage;
-    this.maxBytesInMemory = maxBytesInMemory;
+  public Future(Configuration config, Class<T> clazz, byte[] encoded) {
+    this.config = config;
     this.encoded = encoded;
+    this.clazz = clazz;
   }
 
-  public Future(Storage storage, long maxBytesInMemory, String url) {
+  public Future(Configuration config, Class<T> clazz, String url) {
     this.url = url;
-    this.storage = storage;
-    this.maxBytesInMemory = maxBytesInMemory;
+    this.config = config;
+    this.clazz = clazz;
   }
 
-  public Future(Storage storage, long maxBytesInMemory) {
-    if (storage == null) {
-      throw new IllegalArgumentException("storage can't be null");
+  public Future(T obj, Configuration configuration) throws IOException {
+    byte[] bytes;
+    try {
+      bytes = configuration.getDataConverter().toData(obj);
+    } catch (DataConverterException e) {
+      throw new IOException(e);
     }
 
-    this.storage = storage;
-    this.maxBytesInMemory = maxBytesInMemory;
+    this.config = configuration;
+    this.clazz = (Class<T>) obj.getClass();
+    if (bytes.length <= configuration.getMaxBytes()) {
+      this.encoded = bytes;
+    } else {
+      this.url = configuration.getStorage().put(bytes);
+    }
   }
 
-  public byte[] get() throws IOException {
+  public T get() throws IOException {
     if (encoded != null) {
-      return encoded;
+      return config.getDataConverter().fromData(encoded, clazz, clazz);
     }
 
     if (url != null) {
-      return storage.get(url);
+      return config .getDataConverter().fromData(config.getStorage().get(url), clazz, clazz);
     }
 
     return null;
   }
 
-  public void put(String url, byte[] bytes) throws IOException {
-    this.url = url;
-    if (bytes.length < maxBytesInMemory) {
-      this.encoded = bytes;
-    } else {
-      storage.put(url, bytes);
-    }
-  }
-
   public void delete() throws IOException {
     if (this.encoded == null) {
-      storage.delete(url);
+      config.getStorage().delete(url);
     } else {
       this.encoded = null;
     }

src/main/java/com/uber/cadence/largeblob/Storage.java

@@ -16,6 +16,7 @@
 package com.uber.cadence.largeblob;
 
 import java.io.IOException;
+import java.time.Duration;
 
 /**
  * Storage is an abstraction for storing large parameters to access inside of activities.0
@@ -32,11 +33,26 @@ public interface Storage {
 
   /**
    * Stores data based on uri provided.
-   * @param uri uri.
    * @param bytes bytes.
    * @throws IOException should be thrown in any implementation class in case of problems with the datastore
    */
-  void put(String uri, byte[] bytes) throws IOException;
+  String put(byte[] bytes) throws IOException;
+
+  /**
+   * Stores data based on uri provided.
+   * @param bytes bytes.
+   * @param ttl ttl is used for storages like s3 to define the total time to store the object.
+   * @throws IOException should be thrown in any implementation class in case of problems with the datastore
+   */
+  String put(byte[] bytes, Duration ttl) throws IOException;
+
+  /**
+   * Stores data based on uri provided.
+   * @param key of the data.
+   * @param bytes bytes.
+   * @throws IOException should be thrown in any implementation class in case of problems with the datastore
+   */
+  String put(String key, byte[] bytes) throws IOException;
 
   /**
    * Deletes data based on uri provided.

src/main/java/com/uber/cadence/largeblob/impl/InMemoryStorage.java

@@ -17,8 +17,10 @@
 
 import com.uber.cadence.largeblob.Storage;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 
 public class InMemoryStorage implements Storage {
 
@@ -30,8 +32,21 @@ public byte[] get(String uri) throws IOException {
   }
 
   @Override
-  public void put(String uri, byte[] bytes) throws IOException {
-    storage.put(uri, bytes);
+  public String put(byte[] bytes) throws IOException {
+    String uuid = UUID.randomUUID().toString();
+    storage.put(uuid, bytes);
+    return uuid;
+  }
+
+  @Override
+  public String put(byte[] bytes, Duration ttl) throws IOException {
+    return put(bytes);
+  }
+
+  @Override
+  public String put(String key, byte[] bytes) throws IOException {
+      storage.put(key, bytes);
+      return key;
   }
 
   @Override

src/test/java/com/uber/cadence/internal/largeblob/FutureTest.java

@@ -15,6 +15,7 @@
 
 package com.uber.cadence.internal.largeblob;
 
+import com.uber.cadence.largeblob.Configuration;
 import com.uber.cadence.largeblob.Future;
 import com.uber.cadence.largeblob.Storage;
 import com.uber.cadence.largeblob.impl.InMemoryStorage;
@@ -25,46 +26,42 @@
 import java.nio.charset.StandardCharsets;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
 @RunWith(MockitoJUnitRunner.class)
 public class FutureTest {
 
-  private Storage storage = new InMemoryStorage();
+  private final Storage storage = new InMemoryStorage();
 
   @Test
   public void testPutSmallElement() throws Exception {
-    Future future = new Future(storage, 10);
+    Configuration config = Configuration.newBuilder().setMaxBytes(20L).setStorage(storage).build();
+    Future<String> future = new Future<String>("testValue", config);
 
-    String testValue = "testValue";
-    future.put("test", testValue.getBytes(StandardCharsets.UTF_8));
-
-    assertEquals("testValue", new String(future.get(), StandardCharsets.UTF_8));
-    assertNull(storage.get("test"));
+    assertEquals("testValue", future.get());
+    assertNull(future.getUrl());
   }
 
   @Test
   public void testLargerValuesGetPutInStorage() throws Exception {
-    Future future = new Future(storage, 10);
-
+    Configuration config = Configuration.newBuilder().setMaxBytes(20L).setStorage(storage).build();
     String testValue = "testValuetestValuetestValue";
-    future.put("test", testValue.getBytes(StandardCharsets.UTF_8));
+    Future future = new Future(testValue, config);
 
-    assertEquals("testValuetestValuetestValue", new String(future.get(), StandardCharsets.UTF_8));
-    assertEquals(
-        "testValuetestValuetestValue", new String(storage.get("test"), StandardCharsets.UTF_8));
+    assertEquals("testValuetestValuetestValue", future.get());
+    assertNotNull(storage.get(future.getUrl()));
   }
 
   @Test
   public void testDeleteValueFromStorage() throws Exception {
-    Future future = new Future(storage, 10);
-
+    Configuration config = Configuration.newBuilder().setMaxBytes(20L).setStorage(storage).build();
     String testValue = "testValuetestValuetestValue";
-    future.put("test", testValue.getBytes(StandardCharsets.UTF_8));
+    Future<String> future = new Future<>(testValue, config);
 
-    assertEquals("testValuetestValuetestValue", new String(future.get(), StandardCharsets.UTF_8));
+    assertEquals("testValuetestValuetestValue", future.get());
     assertEquals(
-        "testValuetestValuetestValue", new String(storage.get("test"), StandardCharsets.UTF_8));
+        "testValuetestValuetestValue", config.getDataConverter().fromData(storage.get(future.getUrl()), String.class, String.class));
 
     future.delete();
 
@@ -75,33 +72,33 @@ public void testDeleteValueFromStorage() throws Exception {
 
   @Test
   public void testSmallValueIsDelete() throws Exception {
-    Future future = new Future(storage, 10);
-
-    String testValue = "testValue";
-    future.put("test", testValue.getBytes(StandardCharsets.UTF_8));
+    Configuration config = Configuration.newBuilder().setMaxBytes(20L).setStorage(storage).build();
+    Future<String> future = new Future<>("testValue", config);
 
-    assertEquals("testValue", new String(future.get(), StandardCharsets.UTF_8));
-    assertNull(storage.get("test"));
+    assertEquals("testValue", future.get());
+    assertNull(storage.get(future.getUrl()));
 
     future.delete();
     assertNull(future.get());
   }
 
   @Test
   public void getWorksWhenInitialisingWithEncodedData() throws Exception {
-    Future future = new Future(storage, 10, "testValue".getBytes(StandardCharsets.UTF_8));
+    Configuration configuration = Configuration.newBuilder().setStorage(storage).build();
+    Future<String> future = new Future<>(configuration, String.class, "testValue".getBytes(StandardCharsets.UTF_8));
 
-    assertEquals("testValue", new String(future.get(), StandardCharsets.UTF_8));
+    assertEquals("testValue", future.get());
   }
 
   @Test
   public void getWorksWhenInitialisingWithUrl() throws Exception {
-    Future future = new Future(storage, 10, "test");
+    Configuration configuration = Configuration.newBuilder().setStorage(storage).build();
+    Future<String> future = new Future<>(configuration, String.class, "test");
 
     assertNull(future.get());
 
     storage.put("test", "testValue".getBytes(StandardCharsets.UTF_8));
 
-    assertEquals("testValue", new String(future.get(), StandardCharsets.UTF_8));
+    assertEquals("testValue", future.get());
   }
 }