Allow store offset on consumer closing

When flushing with auto tracking strategy.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/StreamConsumer.java

@@ -352,7 +352,7 @@ void start() {
 
   @Override
   public void store(long offset) {
-    checkOpen();
+    checkNotClosed();
     trackingCallback.accept(offset);
     if (canTrack()) {
       if (offsetBefore(this.lastRequestedStoredOffset, offset)
@@ -444,8 +444,8 @@ boolean sacActive() {
   }
 
   boolean canTrack() {
-    // FIXME check the condition to be able to track
-    return ((this.state() == OPENING || this.state() == OPEN)
+    // closing is OK e.g. when flushing on closing
+    return ((this.state() == OPENING || this.state() == OPEN || this.state() == CLOSING)
             || (this.trackingClient == null && this.state() == RECOVERING))
         && this.name != null;
   }
@@ -526,7 +526,7 @@ void running() {
   }
 
   long storedOffset(Supplier<Client> clientSupplier) {
-    checkOpen();
+    checkNotClosed();
     if (canTrack()) {
       return OffsetTrackingUtils.storedOffset(clientSupplier, this.name, this.stream);
     } else if (this.name == null) {
@@ -607,4 +607,11 @@ void markRecovering() {
   void markOpen() {
     state(OPEN);
   }
+
+  private void checkNotClosed() {
+    if (state() == CLOSED) {
+      // will throw the appropriate exception
+      checkOpen();
+    }
+  }
 }

src/test/java/com/rabbitmq/stream/impl/StreamConsumerTest.java

@@ -971,7 +971,7 @@ void methodsShouldThrowExceptionWhenConsumerIsClosed() {
     ThrowingCallable[] calls =
         new ThrowingCallable[] {() -> consumer.store(1), () -> consumer.storedOffset()};
     Arrays.stream(calls)
-        .forEach(call -> assertThatThrownBy(call).isInstanceOf(IllegalStateException.class));
+        .forEach(call -> assertThatThrownBy(call).isInstanceOf(ResourceClosedException.class));
   }
 
   @Test