a

Author: sgbalogh

app/src/main/java/org/example/app/AccountDemo.java

@@ -8,6 +8,7 @@
 import s2.config.Config;
 import s2.config.Endpoints;
 import s2.types.Age;
+import s2.types.BasinConfig;
 import s2.types.CreateBasinRequest;
 import s2.types.ListBasinsRequest;
 import s2.types.StorageClass;
@@ -33,11 +34,14 @@ public static void main(String[] args) throws Exception {
               .createBasin(
                   CreateBasinRequest.newBuilder()
                       .withBasin(UUID.randomUUID().toString())
-                      .withDefaultStreamConfig(
-                          StreamConfig.newBuilder()
-                              .withRetentionPolicy(new Age(Duration.ofDays(7)))
-                              .withStorageClass(StorageClass.STANDARD)
-                              .build())
+                      .withBasinConfig(
+                          new BasinConfig(
+                              StreamConfig.newBuilder()
+                                  .withRetentionPolicy(new Age(Duration.ofDays(7)))
+                                  .withStorageClass(StorageClass.STANDARD)
+                                  .build(),
+                              false,
+                              false))
                       .build())
               .get();
       logger.info("newBasin={}", newBasin);

s2-sdk/src/main/java/s2/client/Client.java

@@ -20,7 +20,6 @@
 import s2.types.CreateBasinRequest;
 import s2.types.Paginated;
 import s2.types.ReconfigureBasinRequest;
-import s2.types.StreamConfig;
 import s2.v1alpha.AccountServiceGrpc;
 import s2.v1alpha.DeleteBasinRequest;
 import s2.v1alpha.GetBasinConfigRequest;
@@ -125,9 +124,7 @@ public ListenableFuture<BasinConfig> reconfigureBasin(ReconfigureBasinRequest re
                 withStaticRetries(
                     config.maxRetries,
                     () -> this.futureStub.reconfigureBasin(reconfigure.toProto())),
-                resp ->
-                    new BasinConfig(
-                        StreamConfig.fromProto(resp.getConfig().getDefaultStreamConfig())),
+                resp -> BasinConfig.fromProto(resp.getConfig()),
                 executor));
   }
 
@@ -146,9 +143,7 @@ public ListenableFuture<BasinConfig> getBasinConfig(String basin) {
                     () ->
                         this.futureStub.getBasinConfig(
                             GetBasinConfigRequest.newBuilder().setBasin(basin).build())),
-                resp ->
-                    new BasinConfig(
-                        StreamConfig.fromProto(resp.getConfig().getDefaultStreamConfig())),
+                resp -> BasinConfig.fromProto(resp.getConfig()),
                 executor));
   }
 

s2-sdk/src/main/java/s2/client/ManagedAppendSession.java

@@ -374,6 +374,8 @@ public ListenableFuture<Void> closeGracefully() {
     return daemon;
   }
 
+  interface Notification {}
+
   static class InflightRecord {
     final AppendInput input;
     final long entryNanos;
@@ -396,8 +398,6 @@ static InflightRecord construct(AppendInput input, Long meteredBytes) {
     }
   }
 
-  interface Notification {}
-
   class Ack implements Notification {
     final AppendOutput output;
 

s2-sdk/src/main/java/s2/client/ManagedReadSession.java

@@ -101,7 +101,6 @@ public void close() throws Exception {
   }
 
   interface ReadItem {}
-  ;
 
   static class DataItem implements ReadItem {
     final ReadOutput readOutput;

s2-sdk/src/main/java/s2/types/BasinConfig.java

@@ -2,14 +2,28 @@
 
 public class BasinConfig {
   public final StreamConfig defaultStreamConfig;
+  public final boolean createStreamOnAppend;
+  public final boolean createStreamOnRead;
 
-  public BasinConfig(StreamConfig defaultStreamConfig) {
+  public BasinConfig(
+      StreamConfig defaultStreamConfig, boolean createStreamOnAppend, boolean createStreamOnRead) {
     this.defaultStreamConfig = defaultStreamConfig;
+    this.createStreamOnAppend = createStreamOnAppend;
+    this.createStreamOnRead = createStreamOnRead;
+  }
+
+  public static BasinConfig fromProto(s2.v1alpha.BasinConfig basinConfig) {
+    return new BasinConfig(
+        StreamConfig.fromProto(basinConfig.getDefaultStreamConfig()),
+        basinConfig.getCreateStreamOnAppend(),
+        basinConfig.getCreateStreamOnRead());
   }
 
   public s2.v1alpha.BasinConfig toProto() {
     return s2.v1alpha.BasinConfig.newBuilder()
         .setDefaultStreamConfig(defaultStreamConfig.toProto())
+        .setCreateStreamOnAppend(createStreamOnAppend)
+        .setCreateStreamOnRead(createStreamOnRead)
         .build();
   }
 }

s2-sdk/src/main/java/s2/types/CreateBasinRequest.java

@@ -56,8 +56,8 @@ public CreateBasinRequestBuilder withBasin(String basin) {
       return this;
     }
 
-    public CreateBasinRequestBuilder withDefaultStreamConfig(StreamConfig config) {
-      this.config = Optional.of(new BasinConfig(config));
+    public CreateBasinRequestBuilder withBasinConfig(BasinConfig config) {
+      this.config = Optional.of(config);
       return this;
     }
 
@@ -75,7 +75,7 @@ public CreateBasinRequest build() {
       this.basin.ifPresent(BasinUtils::validateBasinName);
       return new CreateBasinRequest(
           this.basin,
-          this.config.orElse(new BasinConfig(StreamConfig.newBuilder().build())),
+          this.config.orElse(new BasinConfig(StreamConfig.newBuilder().build(), false, false)),
           this.assignment);
     }
   }

s2-sdk/src/main/java/s2/types/Header.java

@@ -11,11 +11,11 @@ public Header(ByteString name, ByteString value) {
     this.value = value;
   }
 
-  public s2.v1alpha.Header toProto() {
-    return s2.v1alpha.Header.newBuilder().setName(name).setValue(value).build();
-  }
-
   public static Header fromProto(s2.v1alpha.Header protoHeader) {
     return new Header(protoHeader.getName(), protoHeader.getValue());
   }
+
+  public s2.v1alpha.Header toProto() {
+    return s2.v1alpha.Header.newBuilder().setName(name).setValue(value).build();
+  }
 }

s2-sdk/src/main/java/s2/types/ReadOutput.java

@@ -10,4 +10,4 @@ static ReadOutput fromProto(s2.v1alpha.ReadOutput readOutput) {
     }
     throw new IllegalStateException("Unrecognized readOutput case: " + readOutput);
   }
-}
+}

s2-sdk/src/main/java/s2/types/ReadSessionRequest.java

@@ -24,11 +24,12 @@ public ReadSessionRequest update(Start start, long consumedRecords, long consume
   }
 
   public s2.v1alpha.ReadSessionRequest toProto(String streamName) {
-    s2.v1alpha.ReadSessionRequest.Builder builder = s2.v1alpha.ReadSessionRequest.newBuilder()
-        .setStream(streamName)
-        .setLimit(readLimit.toProto())
-        .setHeartbeats(heartbeats);
-    
+    s2.v1alpha.ReadSessionRequest.Builder builder =
+        s2.v1alpha.ReadSessionRequest.newBuilder()
+            .setStream(streamName)
+            .setLimit(readLimit.toProto())
+            .setHeartbeats(heartbeats);
+
     if (start instanceof Start.SeqNum) {
       builder.setSeqNum(((Start.SeqNum) start).value);
     } else if (start instanceof Start.Timestamp) {
@@ -38,7 +39,7 @@ public s2.v1alpha.ReadSessionRequest toProto(String streamName) {
     } else {
       throw new IllegalStateException("Unknown Start type: " + start.getClass().getName());
     }
-    
+
     return builder.build();
   }
 
@@ -64,9 +65,9 @@ public ReadSessionRequestBuilder withHeartbeats(boolean heartbeats) {
 
     public ReadSessionRequest build() {
       return new ReadSessionRequest(
-          this.start.orElseGet(() -> Start.seqNum(0L)), 
-          this.readLimit.orElse(ReadLimit.NONE), 
+          this.start.orElseGet(() -> Start.seqNum(0L)),
+          this.readLimit.orElse(ReadLimit.NONE),
           this.heartbeats);
     }
   }
-}
+}

s2-sdk/src/main/java/s2/types/SequencedRecord.java

@@ -18,14 +18,6 @@ public class SequencedRecord implements MeteredBytes {
     this.timestamp = timestamp;
   }
 
-  @Override
-  public long meteredBytes() {
-    return 8
-        + (2L * this.headers.size())
-        + this.headers.stream().map(h -> h.name.size() + h.value.size()).reduce(0, Integer::sum)
-        + this.body.size();
-  }
-
   public static SequencedRecord fromProto(s2.v1alpha.SequencedRecord sequencedRecord) {
     return new SequencedRecord(
         sequencedRecord.getSeqNum(),
@@ -35,4 +27,12 @@ public static SequencedRecord fromProto(s2.v1alpha.SequencedRecord sequencedReco
         sequencedRecord.getBody(),
         Instant.ofEpochMilli(sequencedRecord.getTimestamp()));
   }
+
+  @Override
+  public long meteredBytes() {
+    return 8
+        + (2L * this.headers.size())
+        + this.headers.stream().map(h -> h.name.size() + h.value.size()).reduce(0, Integer::sum)
+        + this.body.size();
+  }
 }