Make explicit that an instance has been closed

Fixes #169

Author: acogoluegnes

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

@@ -330,6 +330,7 @@ void start() {
 
   @Override
   public void store(long offset) {
+    checkNotClosed();
     trackingCallback.accept(offset);
     if (canTrack()) {
       if (offsetBefore(this.lastRequestedStoredOffset, offset)
@@ -475,6 +476,7 @@ void running() {
 
   @Override
   public long storedOffset() {
+    checkNotClosed();
     if (canTrack()) {
       // the client can be null by now, so we catch any exception
       QueryOffsetResponse response;
@@ -561,4 +563,10 @@ public String toString() {
         + (trackingClient == null ? "null" : ("\"" + trackingClient.connectionName() + "\""))
         + "}";
   }
+
+  private void checkNotClosed() {
+    if (this.status == Status.CLOSED) {
+      throw new IllegalStateException("This producer instance has been closed");
+    }
+  }
 }

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

@@ -371,11 +371,13 @@ void maybeInitializeLocator() {
 
   @Override
   public StreamCreator streamCreator() {
+    checkNotClosed();
     return new StreamStreamCreator(this);
   }
 
   @Override
   public void deleteStream(String stream) {
+    checkNotClosed();
     this.maybeInitializeLocator();
     Client.Response response = this.locator().delete(stream);
     if (!response.isOk()) {
@@ -391,6 +393,7 @@ public void deleteStream(String stream) {
 
   @Override
   public StreamStats queryStreamStats(String stream) {
+    checkNotClosed();
     StreamStatsResponse response =
         locatorOperation(
             client -> {
@@ -454,6 +457,7 @@ public long committedChunkId() {
 
   @Override
   public ProducerBuilder producerBuilder() {
+    checkNotClosed();
     return new StreamProducerBuilder(this);
   }
 
@@ -475,6 +479,7 @@ void removeConsumer(StreamConsumer consumer) {
 
   @Override
   public ConsumerBuilder consumerBuilder() {
+    checkNotClosed();
     return new StreamConsumerBuilder(this);
   }
 
@@ -734,4 +739,10 @@ public LocatorNotAvailableException() {
       super("Locator not available");
     }
   }
+
+  private void checkNotClosed() {
+    if (this.closed.get()) {
+      throw new IllegalStateException("This environment instance has been closed");
+    }
+  }
 }

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

@@ -268,6 +268,7 @@ public MessageBuilder messageBuilder() {
 
   @Override
   public long getLastPublishingId() {
+    checkNotClosed();
     if (this.name != null && !this.name.isEmpty()) {
       if (canSend()) {
         try {
@@ -516,4 +517,10 @@ public String toString() {
         + (client == null ? "null" : ("\"" + client.connectionName() + "\""))
         + "}";
   }
+
+  private void checkNotClosed() {
+    if (this.closed.get()) {
+      throw new IllegalStateException("This producer instance has been closed");
+    }
+  }
 }

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

@@ -57,6 +57,7 @@
 import java.util.function.UnaryOperator;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
+import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -880,4 +881,17 @@ void offsetZeroShouldBeStored() throws Exception {
     consumer.store(0);
     waitAtMost(() -> consumer.storedOffset() == 0);
   }
+
+  @Test
+  void methodsShouldThrowExceptionWhenConsumerIsClosed() {
+    Consumer consumer =
+        environment.consumerBuilder().stream(stream)
+            .messageHandler((context, message) -> {})
+            .build();
+    consumer.close();
+    ThrowingCallable[] calls =
+        new ThrowingCallable[] {() -> consumer.store(1), () -> consumer.storedOffset()};
+    Arrays.stream(calls)
+        .forEach(call -> assertThatThrownBy(call).isInstanceOf(IllegalStateException.class));
+  }
 }

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

@@ -54,6 +54,7 @@
 import java.net.ConnectException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -67,6 +68,7 @@
 import java.util.stream.IntStream;
 import javax.net.ssl.SNIHostName;
 import javax.net.ssl.SSLParameters;
+import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
@@ -544,4 +546,20 @@ void queryStreamStatsShouldThrowExceptionWhenStreamDoesNotExist() {
           .isInstanceOf(StreamDoesNotExistException.class);
     }
   }
+
+  @Test
+  void methodsShouldThrowExceptionWhenEnvironmentIsClosed() {
+    Environment env = environmentBuilder.build();
+    env.close();
+    ThrowingCallable[] calls =
+        new ThrowingCallable[] {
+          () -> env.streamCreator(),
+          () -> env.producerBuilder(),
+          () -> env.consumerBuilder(),
+          () -> env.deleteStream("does not matter"),
+          () -> env.queryStreamStats("does not matter")
+        };
+    Arrays.stream(calls)
+        .forEach(call -> assertThatThrownBy(call).isInstanceOf(IllegalStateException.class));
+  }
 }

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

@@ -18,6 +18,7 @@
 import static com.rabbitmq.stream.impl.TestUtils.streamName;
 import static com.rabbitmq.stream.impl.TestUtils.waitAtMost;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import ch.qos.logback.classic.Level;
 import com.rabbitmq.stream.BackOffDelayPolicy;
@@ -619,4 +620,12 @@ void subEntryBatchesSentCompressedShouldBeConsumedProperly() {
     publishedBodies.forEach(
         publishBody -> assertThat(consumedBodies.contains(publishBody)).isTrue());
   }
+
+  @Test
+  void methodsShouldThrowExceptionWhenProducerIsClosed() {
+    Producer producer = environment.producerBuilder().stream(stream).build();
+    producer.close();
+    assertThatThrownBy(() -> producer.getLastPublishingId())
+        .isInstanceOf(IllegalStateException.class);
+  }
 }