public CompletionStage<Void> reactiveInitializeCollection(PersistentCollection collection, boolean writing) {
	checkOpenOrWaitingForAutoClose();
	pulseTransactionCoordinator();
	InitializeCollectionEvent event = new InitializeCollectionEvent( collection, this );
	return fire( event, EventType.INIT_COLLECTION,
			(DefaultReactiveInitializeCollectionEventListener l) -> l::onReactiveInitializeCollection )
			.handle( (v, e) -> {
				delayedAfterCompletion();

				if ( e instanceof MappingException ) {
					throw getExceptionConverter().convert( new IllegalArgumentException( e.getMessage() ) );
				}
				else if ( e instanceof RuntimeException ) {
					throw getExceptionConverter().convert( (RuntimeException) e );
				}
				return CompletionStages.returnNullorRethrow( e );
			} );
}
 

/**
 * Asynchronously fetch an association that's configured for lazy loading.
 *
 * <pre>
 * {@code Mutiny.fetch(author.getBook()).map(book -> print(book.getTitle()));}
 * </pre>
 *
 * @param association a lazy-loaded association
 *
 * @return the fetched association, via a {@code Uni}
 *
 * @see org.hibernate.Hibernate#initialize(Object)
 */
static <T> Uni<T> fetch(T association) {
	if ( association == null ) {
		return Uni.createFrom().nullItem();
	}

	SharedSessionContractImplementor session;
	if ( association instanceof HibernateProxy) {
		session = ( (HibernateProxy) association ).getHibernateLazyInitializer().getSession();
	}
	else if ( association instanceof PersistentCollection) {
		session = ( (AbstractPersistentCollection) association ).getSession();
	}
	else {
		return Uni.createFrom().item( association );
	}
	if (session==null) {
		throw new LazyInitializationException("session closed");
	}
	return Uni.createFrom().completionStage(
			( (ReactiveSession) session ).reactiveFetch( association, false )
	);
}
 

/**
 * Asynchronously fetch an association that's configured for lazy loading.
 *
 * <pre>
 * {@code Stage.fetch(author.getBook()).thenAccept(book -> print(book.getTitle()));}
 * </pre>
 *
 * @param association a lazy-loaded association
 *
 * @return the fetched association, via a {@code CompletionStage}
 *
 * @see org.hibernate.Hibernate#initialize(Object)
 */
static <T> CompletionStage<T> fetch(T association) {
	if ( association == null ) {
		return CompletionStages.nullFuture();
	}

	SharedSessionContractImplementor session;
	if ( association instanceof HibernateProxy) {
		session = ( (HibernateProxy) association ).getHibernateLazyInitializer().getSession();
	}
	else if ( association instanceof PersistentCollection) {
		session = ( (AbstractPersistentCollection) association ).getSession();
	}
	else {
		return CompletionStages.completedFuture( association );
	}
	if (session==null) {
		throw new LazyInitializationException("session closed");
	}
	return ( (ReactiveSession) session ).reactiveFetch( association, false );
}
 

private void evictCollection(PersistentCollection collection) {
	CollectionEntry ce = (CollectionEntry) getSession().getPersistenceContext().getCollectionEntries().remove(collection);
	if ( LOG.isDebugEnabled() ) {
		LOG.debugf(
				"Evicting collection: %s",
				MessageHelper.collectionInfoString( ce.getLoadedPersister(),
						collection,
						ce.getLoadedKey(),
						getSession() ) );
	}
	if (ce.getLoadedPersister() != null && ce.getLoadedPersister().getBatchSize() > 1) {
		getSession().getPersistenceContext().getBatchFetchQueue().removeBatchLoadableCollection(ce);
	}
	if ( ce.getLoadedPersister() != null && ce.getLoadedKey() != null ) {
		//TODO: is this 100% correct?
		getSession().getPersistenceContext().getCollectionsByKey().remove(
				new CollectionKey( ce.getLoadedPersister(), ce.getLoadedKey() )
		);
	}
}
 

/**
 * Add a collection to the cache, with a given collection entry.
 *
 * @param coll The collection for which we are adding an entry.
 * @param entry The entry representing the collection.
 * @param key The key of the collection's entry.
 */
private void addCollection(PersistentCollection coll, CollectionEntry entry, Serializable key) {
	collectionEntries.put( coll, entry );
	final CollectionKey collectionKey = new CollectionKey( entry.getLoadedPersister(), key );
	final PersistentCollection old = collectionsByKey.put( collectionKey, coll );
	if ( old != null ) {
		if ( old == coll ) {
			throw new AssertionFailure( "bug adding collection twice" );
		}
		// or should it actually throw an exception?
		old.unsetSession( session );
		collectionEntries.remove( old );
		// watch out for a case where old is still referenced
		// somewhere in the object graph! (which is a user error)
	}
}
 

@Override
public String toLoggableString(Object value, SessionFactoryImplementor factory)
		throws HibernateException {
	if ( value == null ) {
		return "null";
	}

	if ( !getReturnedClass().isInstance( value ) && !PersistentCollection.class.isInstance( value ) ) {
		// its most likely the collection-key
		final CollectionPersister persister = getPersister( factory );
		if ( persister.getKeyType().getReturnedClass().isInstance( value ) ) {
			return getRole() + "#" + getPersister( factory ).getKeyType().toLoggableString( value, factory );
		}
		else {
			// although it could also be the collection-id
			if ( persister.getIdentifierType() != null
					&& persister.getIdentifierType().getReturnedClass().isInstance( value ) ) {
				return getRole() + "#" + getPersister( factory ).getIdentifierType().toLoggableString( value, factory );
			}
		}
	}
	return renderLoggableString( value, factory );
}
 

private static void processNeverReferencedCollection(PersistentCollection coll, SessionImplementor session)
		throws HibernateException {
	final PersistenceContext persistenceContext = session.getPersistenceContext();
	final CollectionEntry entry = persistenceContext.getCollectionEntry( coll );

	if ( LOG.isDebugEnabled() ) {
		LOG.debugf(
				"Found collection with unloaded owner: %s",
				MessageHelper.collectionInfoString( 
						entry.getLoadedPersister(),
						coll,
						entry.getLoadedKey(),
						session
				)
		);
	}

	entry.setCurrentPersister( entry.getLoadedPersister() );
	entry.setCurrentKey( entry.getLoadedKey() );

	prepareCollectionForUpdate( coll, entry, session.getFactory() );

}
 

public PreCollectionRemoveEvent(CollectionPersister collectionPersister,
								PersistentCollection collection,
								EventSource source,
								Object loadedOwner) {
	super( collectionPersister, collection, source,
			loadedOwner,
			getOwnerIdOrNull( loadedOwner, source ) );
}
 

/**
 * Check if the proxy or persistent collection is initialized.
 *
 * @param proxy a persistable object, proxy, persistent collection or <tt>null</tt>
 * @return true if the argument is already initialized, or is not a proxy or collection
 */
@SuppressWarnings("SimplifiableIfStatement")
public static boolean isInitialized(Object proxy) {
	if ( proxy instanceof HibernateProxy ) {
		return !( (HibernateProxy) proxy ).getHibernateLazyInitializer().isUninitialized();
	}
	else if ( proxy instanceof PersistentCollection ) {
		return ( (PersistentCollection) proxy ).wasInitialized();
	}
	else {
		return true;
	}
}
 

@Override
public void accept(PersistentCollection nonLazyCollection) {
	stage = stage.thenCompose(
			v -> ( (ReactiveSession) getSession() )
					.reactiveFetch( nonLazyCollection, true )
					.thenAccept(vv -> {})
	);
}
 

public PostCollectionRecreateEvent( CollectionPersister collectionPersister,
									PersistentCollection collection,
									EventSource source ) {
	super( collectionPersister, collection, source,
			collection.getOwner(),
			getOwnerIdOrNull( collection.getOwner(), source ) );
}
 

@Override
public void addInitializedDetachedCollection(CollectionPersister collectionPersister, PersistentCollection collection)
		throws HibernateException {
	if ( collection.isUnreferenced() ) {
		//treat it just like a new collection
		addCollection( collection, collectionPersister );
	}
	else {
		final CollectionEntry ce = new CollectionEntry( collection, session.getFactory() );
		addCollection( collection, ce, collection.getKey() );
	}
}
 

/**
 * If object is proxy, get unwrapped non-proxy object.
 *
 * @param proxy Object to check and unwrap
 * @return Unwrapped object if proxyied, if not just returns same object
 */
@SuppressWarnings( "unchecked" )
public static <T> T unwrap( T proxy )
{
    if ( !isProxy( proxy ) )
    {
        return proxy;
    }

    Hibernate.initialize( proxy );

    if ( HibernateProxy.class.isInstance( proxy ) )
    {
        Object result = ((HibernateProxy) proxy).writeReplace();

        if ( !SerializableProxy.class.isInstance( result ) )
        {
            return (T) result;
        }
    }

    if ( PersistentCollection.class.isInstance( proxy ) )
    {
        PersistentCollection persistentCollection = (PersistentCollection) proxy;

        if ( PersistentSet.class.isInstance( persistentCollection ) )
        {
            Map<?, ?> map = (Map<?, ?>) persistentCollection.getStoredSnapshot();
            return (T) new LinkedHashSet<>( map.keySet() );
        }

        return (T) persistentCollection.getStoredSnapshot();
    }

    return proxy;
}
 

@Override
public void updateRows(PersistentCollection collection, Serializable id, SharedSessionContractImplementor session)
		throws HibernateException {

	if ( !isInverse && collection.isRowUpdatePossible() ) {

		LOG.debugf( "Updating rows of collection: %s#%s", navigableRole.getFullPath(), id );

		// update all the modified entries
		int count = doUpdateRows( id, collection, session );

		LOG.debugf( "Done updating rows: %s updated", count );
	}
}
 

/**
 * Attempt to locate the loading collection given the owner's key.  The lookup here
 * occurs against all result-set contexts...
 *
 * @param persister The collection persister
 * @param ownerKey The owner key
 * @return The loading collection, or null if not found.
 */
public PersistentCollection locateLoadingCollection(CollectionPersister persister, Serializable ownerKey) {
	final LoadingCollectionEntry lce = locateLoadingCollectionEntry( new CollectionKey( persister, ownerKey ) );
	if ( lce != null ) {
		if ( LOG.isTraceEnabled() ) {
			LOG.tracef(
					"Returning loading collection: %s",
					MessageHelper.collectionInfoString( persister, ownerKey, getSession().getFactory() )
			);
		}
		return lce.getCollection();
	}
	return null;
}
 

@Override
public Object processCollection(Object collection, CollectionType type) throws HibernateException {
	if ( collection == null ) {
		return null;
	}

	final SessionImplementor session = getSession();
	final CollectionPersister persister = session.getFactory().getCollectionPersister( type.getRole() );

	if ( collection instanceof PersistentCollection ) {
		final PersistentCollection persistentCollection = (PersistentCollection) collection;
		if ( persistentCollection.setCurrentSession( session ) ) {
			if ( isOwnerUnchanged( persistentCollection, persister, extractCollectionKeyFromOwner( persister ) ) ) {
				// a "detached" collection that originally belonged to the same entity
				if ( persistentCollection.isDirty() ) {
					throw new HibernateException( "reassociated object has dirty collection" );
				}
				reattachCollection( persistentCollection, type );
			}
			else {
				// a "detached" collection that belonged to a different entity
				throw new HibernateException( "reassociated object has dirty collection reference" );
			}
		}
		else {
			// a collection loaded in the current session
			// can not possibly be the collection belonging
			// to the entity passed to update()
			throw new HibernateException( "reassociated object has dirty collection reference" );
		}
	}
	else {
		// brand new collection
		//TODO: or an array!! we can't lock objects with arrays now??
		throw new HibernateException( "reassociated object has dirty collection reference (or an array)" );
	}

	return null;
}
 

/**
 * Replace the elements of a collection with the elements of another collection.
 *
 * @param original The 'source' of the replacement elements (where we copy from)
 * @param target The target of the replacement elements (where we copy to)
 * @param owner The owner of the collection being merged
 * @param copyCache The map of elements already replaced.
 * @param session The session from which the merge event originated.
 * @return The merged collection.
 */
public Object replaceElements(
		Object original,
		Object target,
		Object owner,
		Map copyCache,
		SharedSessionContractImplementor session) {
	// TODO: does not work for EntityMode.DOM4J yet!
	java.util.Collection result = ( java.util.Collection ) target;
	result.clear();

	// copy elements into newly empty target collection
	Type elemType = getElementType( session.getFactory() );
	Iterator iter = ( (java.util.Collection) original ).iterator();
	while ( iter.hasNext() ) {
		result.add( elemType.replace( iter.next(), null, session, owner, copyCache ) );
	}

	// if the original is a PersistentCollection, and that original
	// was not flagged as dirty, then reset the target's dirty flag
	// here after the copy operation.
	// </p>
	// One thing to be careful of here is a "bare" original collection
	// in which case we should never ever ever reset the dirty flag
	// on the target because we simply do not know...
	if ( original instanceof PersistentCollection ) {
		if ( result instanceof PersistentCollection ) {
			final PersistentCollection originalPersistentCollection = (PersistentCollection) original;
			final PersistentCollection resultPersistentCollection = (PersistentCollection) result;

			preserveSnapshot( originalPersistentCollection, resultPersistentCollection, elemType, owner, copyCache, session );

			if ( ! originalPersistentCollection.isDirty() ) {
				resultPersistentCollection.clearDirty();
			}
		}
	}

	return result;
}
 

@Override
public void addUninitializedCollection(CollectionPersister persister, PersistentCollection collection, Serializable id) {
	final CollectionEntry ce = new CollectionEntry( collection, persister, id, flushing );
	addCollection( collection, ce, id );
	if ( persister.getBatchSize() > 1 ) {
		getBatchFetchQueue().addBatchLoadableCollection( collection, ce );
	}
}
 

/**
 * 1. Recreate the collection key -> collection map
 * 2. rebuild the collection entries
 * 3. call Interceptor.postFlush()
 */
protected void postFlush(SessionImplementor session) throws HibernateException {

	LOG.trace( "Post flush" );

	final PersistenceContext persistenceContext = session.getPersistenceContext();
	persistenceContext.getCollectionsByKey().clear();
	
	// the database has changed now, so the subselect results need to be invalidated
	// the batch fetching queues should also be cleared - especially the collection batch fetching one
	persistenceContext.getBatchFetchQueue().clear();

	for ( Map.Entry<PersistentCollection, CollectionEntry> me : IdentityMap.concurrentEntries( persistenceContext.getCollectionEntries() ) ) {
		CollectionEntry collectionEntry = me.getValue();
		PersistentCollection persistentCollection = me.getKey();
		collectionEntry.postFlush(persistentCollection);
		if ( collectionEntry.getLoadedPersister() == null ) {
			//if the collection is dereferenced, unset its session reference and remove from the session cache
			//iter.remove(); //does not work, since the entrySet is not backed by the set
			persistentCollection.unsetSession( session );
			persistenceContext.getCollectionEntries()
					.remove(persistentCollection);
		}
		else {
			//otherwise recreate the mapping between the collection and its key
			CollectionKey collectionKey = new CollectionKey(
					collectionEntry.getLoadedPersister(),
					collectionEntry.getLoadedKey()
			);
			persistenceContext.getCollectionsByKey().put(collectionKey, persistentCollection);
		}
	}

}
 

@Override
public CollectionEntry addInitializedCollection(CollectionPersister persister, PersistentCollection collection, Serializable id)
		throws HibernateException {
	final CollectionEntry ce = new CollectionEntry( collection, persister, id, flushing );
	ce.postInitialize( collection );
	addCollection( collection, ce, id );
	return ce;
}
 

public PostCollectionRemoveEvent(
		CollectionPersister collectionPersister,
		PersistentCollection collection,
		EventSource source,
		Object loadedOwner) {
	super( collectionPersister, collection, source, loadedOwner, getOwnerIdOrNull( loadedOwner, source ) );
}
 

/**
 * For newly wrapped collections, or dereferenced collection wrappers
 */
public CollectionEntry(CollectionPersister persister, PersistentCollection collection) {
	// new collections that get found + wrapped
	// during flush shouldn't be ignored
	ignore = false;

	collection.clearDirty(); //a newly wrapped collection is NOT dirty (or we get unnecessary version updates)

	snapshot = persister.isMutable() ?
			collection.getSnapshot(persister) :
			null;
	collection.setSnapshot(loadedKey, role, snapshot);
}
 

/**
 * Has the owner of the collection changed since the collection
 * was snapshotted and detached?
 */
protected static boolean isOwnerUnchanged(
		final PersistentCollection snapshot, 
		final CollectionPersister persister, 
		final Serializable id
) {
	return isCollectionSnapshotValid(snapshot) &&
			persister.getRole().equals( snapshot.getRole() ) &&
			id.equals( snapshot.getKey() );
}
 

/**
 * Determine if the collection is "really" dirty, by checking dirtiness
 * of the collection elements, if necessary
 */
private void dirty(PersistentCollection collection) throws HibernateException {

	boolean forceDirty = collection.wasInitialized() &&
			!collection.isDirty() && //optimization
			getLoadedPersister() != null &&
			getLoadedPersister().isMutable() && //optimization
			( collection.isDirectlyAccessible() || getLoadedPersister().getElementType().isMutable() ) && //optimization
			!collection.equalsSnapshot( getLoadedPersister() );

	if ( forceDirty ) {
		collection.dirty();
	}

}
 

public PostCollectionUpdateEvent(
		CollectionPersister collectionPersister,
		PersistentCollection collection,
		EventSource source) {
	super(
			collectionPersister,
			collection,
			source,
			getLoadedOwnerOrNull( collection, source ),
			getLoadedOwnerIdOrNull( collection, source )
	);
}
 

/**
 * Called after a successful flush
 */
public void postFlush(PersistentCollection collection) throws HibernateException {
	if ( isIgnore() ) {
		ignore = false;
	}
	else if ( !isProcessed() ) {
		throw new HibernateException( LOG.collectionNotProcessedByFlush( collection.getRole() ) );
	}
	collection.setSnapshot(loadedKey, role, snapshot);
}
 

/**
 * Reset the stored snapshot for both the persistent collection and this collection entry. 
 * Used during the merge of detached collections.
 * 
 * @param collection the persistentcollection to be updated
 * @param storedSnapshot the new stored snapshot
 */
public void resetStoredSnapshot(PersistentCollection collection, Serializable storedSnapshot) {
	LOG.debugf("Reset storedSnapshot to %s for %s", storedSnapshot, this);

	if ( fromMerge ) {
		return; // EARLY EXIT!
	}

	snapshot = storedSnapshot;
	collection.setSnapshot( loadedKey, role, snapshot );
	fromMerge = true;
}
 

public static Class<?> getRealClass( Class<?> klass )
{
    if ( ProxyFactory.isProxyClass( klass ) )
    {
        klass = klass.getSuperclass();
    }

    while ( PersistentCollection.class.isAssignableFrom( klass ) )
    {
        klass = klass.getSuperclass();
    }

    return klass;
}
 

/**
 * Initialize the flags of the CollectionEntry, including the
 * dirty check.
 */
private void prepareCollectionFlushes(PersistenceContext persistenceContext) throws HibernateException {

	// Initialize dirty flags for arrays + collections with composite elements
	// and reset reached, doupdate, etc.

	LOG.debug( "Dirty checking collections" );

	for ( Map.Entry<PersistentCollection,CollectionEntry> entry :
			IdentityMap.concurrentEntries( (Map<PersistentCollection,CollectionEntry>) persistenceContext.getCollectionEntries() )) {
		entry.getValue().preFlush( entry.getKey() );
	}
}
 

public PreCollectionUpdateEvent(CollectionPersister collectionPersister,
								PersistentCollection collection,
								EventSource source) {
	super( collectionPersister, collection, source,
			getLoadedOwnerOrNull( collection, source ),
			getLoadedOwnerIdOrNull( collection, source ) );
}