fix(batch): improve reconnect and shutdown mechanisms

- change condition for reconnecting -- send will always wait for recv
to complete, so checking send.isDone is meaningless. Instead, we rely
on the context's state and the data we've observer from the server.
- Emit OOM's 'shutdown sequence' via the Recv handle, not via BatchContext::onEvent.
This ensures the recv.done is completed correctly on EOF to unblock send's exit.
- add a protoc guard to Reconnecting -- an EOF event should not arrive in this
state.
- fixed BatchContextTest so that stream is not automatically closed when
client closes it's half

Author: bevzzz

src/main/java/io/weaviate/client6/v1/api/collections/batch/BatchContext.java

@@ -519,7 +519,7 @@ private void trySend() {
             // processed all previous requests; the WIP buffer is empty in that case.
             //
             // It is possible that the server will be restarted or the stream will be
-            // hungup before client receives all Results, in which case we might need
+            // hung up before client receives all Results, in which case we might need
             // to re-submit the items remaining in the WIP buffer.
             recv.get();
 
@@ -610,7 +610,12 @@ private Recv(BatchContext<?> context) {
     @Override
     public void onNext(Event event) {
       try {
-        context.onEvent(event);
+        if (event == Event.EOF) {
+          // Handle synthetic EOF which the Oom state can send to initiate a shutdown.
+          onCompleted();
+        } else {
+          context.onEvent(event);
+        }
       } catch (Exception e) {
         context.onEvent(new Event.ClientError(e));
       }
@@ -748,11 +753,17 @@ private void onShuttingDown() {
 
     private void onStreamClosed(Event event) {
       if (event instanceof Event.StreamHangup hangup) {
+        // TODO(dyma): impprove logging
         hangup.exception().printStackTrace();
       }
-      if (!send.isDone()) {
-        setState(new Reconnecting(maxReconnectRetries));
+
+      // The only time we should not try to reconnect is if the context
+      // is gracefully shutting down after a call to close() and the server
+      // has returned Results for all previous batches.
+      if (closed && wip.isEmpty()) {
+        return;
       }
+      setState(new Reconnecting(maxReconnectRetries));
     }
 
     private void onClientError(Event.ClientError error) {
@@ -790,11 +801,12 @@ private void initiateShutdown() {
       // receive an Event.SHUTTING_DOWN, it would cancel this execution of this
       // very sequence. Instead, we delegate to our parent BaseState which normally
       // handles these events.
+      final Recv events = (Recv) recv;
       if (!Thread.currentThread().isInterrupted()) {
-        BatchContext.this.onEvent(Event.SHUTTING_DOWN);
+        events.onNext(Event.SHUTTING_DOWN);
       }
       if (!Thread.currentThread().isInterrupted()) {
-        BatchContext.this.onEvent(Event.EOF);
+        events.onNext(Event.EOF);
       }
     }
 
@@ -905,7 +917,10 @@ public void onEvent(Event event) {
         } else {
           reconnectAfter(2 ^ retries);
         }
+      } else if (event == Event.EOF) {
+        throw ProtocolViolationException.illegalStateTransition(this, event);
       } else {
+        System.out.println("Event while reconnecting " + event);
         super.onEvent(event);
       }