plugins {
	id 'application'
}
repositories {
	mavenCentral()
}
dependencies {
	implementation 'org.mongodb:mongodb-driver-reactivestreams:4.5.0'
}
application {
	mainClass = 'company.Test'
}

package company;

import com.mongodb.client.model.Filters;
import com.mongodb.client.result.InsertOneResult;
import com.mongodb.reactivestreams.client.MongoClients;

import org.bson.Document;
import org.bson.codecs.Codec;
import org.bson.codecs.configuration.CodecRegistry;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class Test {
	public static class CustomCodecRegistry implements CodecRegistry {
		private final CodecRegistry codecRegistry;

		public CustomCodecRegistry(CodecRegistry codecRegistry) {
			this.codecRegistry = codecRegistry;
		}

		@Override
		public <T> Codec<T> get(Class<T> clazz) {
			return get(clazz, codecRegistry);
		}

		@Override
		public <T> Codec<T> get(Class<T> clazz, CodecRegistry registry) {
			return registry.get(clazz);
		}
	}

	public static void main(String[] args) throws InterruptedException {
		var client = MongoClients.create();

		var collection = client.getDatabase("test").getCollection("collection");

		collection.insertOne(new Document("bad", "42")).subscribe(new Subscriber<>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(1);
			}

			@Override
			public void onNext(final InsertOneResult success) {
			}

			@Override
			public void onError(Throwable t) {
				t.printStackTrace();
			}

			@Override
			public void onComplete() {
				collection
					.withCodecRegistry(new CustomCodecRegistry(collection.getCodecRegistry()))
					.find(Filters.eq("bad", "42"))
					.subscribe(new Subscriber<>() {
						@Override
						public void onSubscribe(Subscription s) {
							s.request(1);
						}

						@Override
						public void onNext(Document t) {
							System.err.println(t);
						}

						@Override
						public void onError(Throwable t) {
							t.printStackTrace();
						}

						@Override
						public void onComplete() {
						}
				});
			}
		});

		Thread.sleep(5000);
	}
}