proposed fix for #707

make sure that queued requests use the right session and entity args

Author: gavinking

hibernate-reactive-core/src/main/java/org/hibernate/reactive/id/impl/BlockingIdentifierGenerator.java

@@ -38,7 +38,27 @@ public abstract class BlockingIdentifierGenerator implements ReactiveIdentifierG
     private int loValue;
     private long hiValue;
 
-    private volatile List<CompletableFuture<Long>> queue = null;
+    private volatile List<Queued> queue = null;
+
+    /**
+     * A queued request for an identifier that can't be satisfied immediately
+     * because we're currently fetching a new "hi" value in a different request.
+     */
+    private final class Queued {
+        private final ReactiveConnectionSupplier session;
+        private final Object entity;
+        private final CompletableFuture<Long> completion;
+
+        private Queued(ReactiveConnectionSupplier session, Object entity, CompletableFuture<Long> completion) {
+            this.session = session;
+            this.entity = entity;
+            this.completion = completion;
+        }
+
+        private void complete() {
+            generate(session, entity).thenAccept(completion::complete);
+        }
+    }
 
     protected synchronized long next() {
         return loValue>0 && loValue<getBlockSize()
@@ -70,7 +90,7 @@ public CompletionStage<Long> generate(ReactiveConnectionSupplier session, Object
                     // go off and fetch the next hi value from db
                     nextHiValue(session).thenAccept( id -> {
 //						Vertx.currentContext().runOnContext(v -> {
-                        List<CompletableFuture<Long>> list;
+                        List<Queued> list;
                         synchronized (this) {
                             // clone ref to the queue
                             list = queue;
@@ -79,14 +99,13 @@ public CompletionStage<Long> generate(ReactiveConnectionSupplier session, Object
                             result.complete( next(id) );
                         }
                         // send waiting streams back to try again
-                        list.forEach( completion -> generate(session, entity)
-                                .thenAccept(completion::complete) );
+                        list.forEach(Queued::complete);
 //						} );
                     } );
                 }
                 else {
                     // wait for the concurrent fetch to complete
-                    queue.add(result);
+                    queue.add( new Queued(session, entity, result) );
                 }
                 return result;
             }