Delay connection to server until necessary

Not just when builders are created, but just before to-be-created
objects are actually instantiated.

References #14

Author: acogoluegnes

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

@@ -87,6 +87,7 @@ public Consumer build() {
       throw new IllegalArgumentException("A name must be set if a tracking strategy is specified");
     }
 
+    this.environment.maybeInitializeLocator();
     TrackingConfiguration trackingConfiguration;
     if (this.autoTrackingStrategy != null) {
       trackingConfiguration =

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

@@ -336,7 +336,7 @@ public ByteBufAllocator byteBufAllocator() {
     return byteBufAllocator;
   }
 
-  private void maybeInitializeLocator() {
+  void maybeInitializeLocator() {
     if (this.locatorInitialized.compareAndSet(false, true)) {
       try {
         this.locatorInitializationSequence.run();
@@ -349,13 +349,12 @@ private void maybeInitializeLocator() {
 
   @Override
   public StreamCreator streamCreator() {
-    maybeInitializeLocator();
     return new StreamStreamCreator(this);
   }
 
   @Override
   public void deleteStream(String stream) {
-    maybeInitializeLocator();
+    this.maybeInitializeLocator();
     Client.Response response = this.locator().delete(stream);
     if (!response.isOk()) {
       throw new StreamException(
@@ -370,7 +369,6 @@ public void deleteStream(String stream) {
 
   @Override
   public ProducerBuilder producerBuilder() {
-    maybeInitializeLocator();
     return new StreamProducerBuilder(this);
   }
 
@@ -392,7 +390,6 @@ void removeConsumer(StreamConsumer consumer) {
 
   @Override
   public ConsumerBuilder consumerBuilder() {
-    maybeInitializeLocator();
     return new StreamConsumerBuilder(this);
   }
 

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

@@ -155,6 +155,7 @@ public Producer build() {
     if (subEntrySize > 1 && compression == null) {
       compression = Compression.NONE;
     }
+    this.environment.maybeInitializeLocator();
     Producer producer;
     if (this.routingKeyExtractor == null) {
       producer =

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

@@ -67,6 +67,7 @@ public void create() {
     if (stream == null) {
       throw new IllegalArgumentException("Stream cannot be null");
     }
+    this.environment.maybeInitializeLocator();
     Client.Response response =
         environment.locator().create(stream, streamParametersBuilder.build());
     if (!response.isOk()

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

@@ -29,12 +29,15 @@
 import com.rabbitmq.stream.ConfirmationHandler;
 import com.rabbitmq.stream.Constants;
 import com.rabbitmq.stream.Consumer;
+import com.rabbitmq.stream.ConsumerBuilder;
 import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.EnvironmentBuilder;
 import com.rabbitmq.stream.Host;
 import com.rabbitmq.stream.Message;
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.Producer;
+import com.rabbitmq.stream.ProducerBuilder;
+import com.rabbitmq.stream.StreamCreator;
 import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import com.rabbitmq.stream.impl.MonitoringTestUtils.EnvironmentInfo;
@@ -456,18 +459,15 @@ void instanciationShouldSucceedWhenLazyInitIsEnabledAndHostIsNotKnown() {
             .lazyInitialization(true)
             .build()) {
 
-      assertThatThrownBy(() -> env.streamCreator().stream("should not have been created").create())
-          .isInstanceOf(StreamException.class);
+      StreamCreator streamCreator = env.streamCreator().stream("should not have been created");
+      assertThatThrownBy(() -> streamCreator.create()).isInstanceOf(StreamException.class);
       assertThatThrownBy(() -> env.deleteStream("should not exist"))
           .isInstanceOf(StreamException.class);
-      assertThatThrownBy(() -> env.producerBuilder().stream(stream).build())
-          .isInstanceOf(StreamException.class);
-      assertThatThrownBy(
-              () ->
-                  env.consumerBuilder().stream(stream)
-                      .messageHandler((context, message) -> {})
-                      .build())
-          .isInstanceOf(StreamException.class);
+      ProducerBuilder producerBuilder = env.producerBuilder().stream(this.stream);
+      assertThatThrownBy(() -> producerBuilder.build()).isInstanceOf(StreamException.class);
+      ConsumerBuilder consumerBuilder =
+          env.consumerBuilder().stream(this.stream).messageHandler((context, message) -> {});
+      assertThatThrownBy(() -> consumerBuilder.build()).isInstanceOf(StreamException.class);
     }
   }