org.junit.jupiter.api.extension.ExtensionConfigurationException#com.mongodb.reactivestreams.client.MongoClient源码实例Demo

public NewsServicePublisher(MongoClient client, String categoryOfInterests) {
	ScheduledPublisher<NewsLetter> scheduler = new ScheduledPublisher<>(
			() -> new NewsPreparationOperator(
					new DBPublisher(
							client.getDatabase("news")
							      .getCollection("news", News.class),
							categoryOfInterests
					),
					"Some Digest"
			),
			1, TimeUnit.DAYS
	);

	SmartMulticastProcessor processor = new SmartMulticastProcessor();
	scheduler.subscribe(processor);

	this.processor = processor;
}
 

@BeforeMethod
public void setUp() {

    map = TestHelper.createMap(true);

    mockRecord = mock(Record.class);
    mockSinkContext = mock(SinkContext.class);
    mockMongoClient = mock(MongoClient.class);
    mockMongoDb = mock(MongoDatabase.class);
    mockMongoColl = mock(MongoCollection.class);
    mockPublisher = mock(Publisher.class);
    sink = new MongoSink(() -> mockMongoClient);


    when(mockMongoClient.getDatabase(anyString())).thenReturn(mockMongoDb);
    when(mockMongoDb.getCollection(anyString())).thenReturn(mockMongoColl);
    when(mockMongoDb.getCollection(anyString()).insertMany(any())).thenReturn(mockPublisher);
}
 

@BeforeMethod
public void setUp() {

    map = TestHelper.createMap(true);

    mockSourceContext = mock(SourceContext.class);
    mockMongoClient = mock(MongoClient.class);
    mockMongoDb = mock(MongoDatabase.class);
    mockMongoColl = mock(MongoCollection.class);
    mockPublisher = mock(ChangeStreamPublisher.class);

    source = new MongoSource(() -> mockMongoClient);

    when(mockMongoClient.getDatabase(anyString())).thenReturn(mockMongoDb);
    when(mockMongoDb.getCollection(anyString())).thenReturn(mockMongoColl);
    when(mockMongoColl.watch()).thenReturn(mockPublisher);
    when(mockPublisher.batchSize(anyInt())).thenReturn(mockPublisher);
    when(mockPublisher.fullDocument(any())).thenReturn(mockPublisher);

    doAnswer((invocation) -> {
        subscriber = invocation.getArgument(0, Subscriber.class);
        return null;
    }).when(mockPublisher).subscribe(any());
}
 

private MongoInstance(MongoClient client, Closer closer) {
  Objects.requireNonNull(closer, "closer");
  this.client = Objects.requireNonNull(client, "client");
  closer.register(client);

  // drop database if exists (to have a clean test)
  if (Flowable.fromPublisher(client.listDatabaseNames()).toList().blockingGet().contains(DBNAME)) {
    Success success = Flowable.fromPublisher(client.getDatabase(DBNAME).drop()).blockingFirst();
  }

  this.database = client.getDatabase(DBNAME);

  closer.register(database::drop);

  this.closer = closer;
}
 

static MongoInstance create() {
  final String uri = System.getProperty("mongo");


  if (uri != null) {
    // connect to remote mongo server
    return new MongoInstance(MongoClients.create(uri));
  }

  final MongoServer server = new MongoServer(new MemoryBackend());
  final InetSocketAddress address = server.bind();
  final Closer closer = Closer.create();
  closer.register(server::shutdownNow);

  final MongoClient client = MongoClients.create(String.format("mongodb://127.0.0.1:%d", address.getPort()));
  return new MongoInstance(client, closer);
}
 

public MongoEventStore(MongoClient mongoClient, MongoConfiguration mongoConfiguration, IEventSerializer eventSerializer) {
    this.mongoClient = mongoClient;
    this.eventSerializer = eventSerializer;
    this.mongoConfiguration = mongoConfiguration;
    this.duplicateCode = mongoConfiguration.getDuplicateCode();
    this.versionIndexName = mongoConfiguration.getEventTableVersionUniqueIndexName();
    this.commandIndexName = mongoConfiguration.getEventTableCommandIdUniqueIndexName();
}
 

@Override
public void initialize(GenericApplicationContext context) {
	context.registerBean(MongoClientSettingsBuilderCustomizer.class, () -> new MongoReactiveAutoConfiguration.NettyDriverConfiguration().nettyDriverCustomizer(context.getDefaultListableBeanFactory().getBeanProvider(MongoClientSettings.class)));

	MongoReactiveAutoConfiguration configuration = new MongoReactiveAutoConfiguration();
	context.registerBean(MongoClient.class, () -> configuration.reactiveStreamsMongoClient(this.properties, context.getEnvironment(), context.getBeanProvider(MongoClientSettingsBuilderCustomizer.class), context.getBeanProvider(MongoClientSettings.class)), (definition) -> {
		if (embeddedServer) {
			definition.setDependsOn("embeddedMongoServer");
		}
	});
}
 

@Override
public void initialize(GenericApplicationContext context) {
	MongoReactiveDataAutoConfiguration configuration = new MongoReactiveDataAutoConfiguration();
	context.registerBean(MappingMongoConverter.class, () -> configuration.mappingMongoConverter(context.getBean(MongoMappingContext.class), context.getBean(MongoCustomConversions.class)));
	context.registerBean(SimpleReactiveMongoDatabaseFactory.class, () -> configuration.reactiveMongoDatabaseFactory(this.properties, context.getBean(MongoClient.class)));
	context.registerBean(ReactiveMongoTemplate.class, () -> configuration.reactiveMongoTemplate(context.getBean(ReactiveMongoDatabaseFactory.class), context.getBean(MongoConverter.class)));
}
 

public RxMongoDriverQueryService(
   MongoClient mongoClient,
   @Value("${spring.data.mongodb.database}") String dbName
) {
   this.mongoClient = mongoClient;
   this.dbName = dbName;
}
 

@Bean
MongoClient mongoClient(MongoProperties properties) {
    ConnectionString connectionString = new ConnectionString(properties.determineUri());
    MongoClientSettings.Builder builder = MongoClientSettings
            .builder()
            .streamFactoryFactory(NettyStreamFactory::new)
            .applyToClusterSettings(b -> b.applyConnectionString(connectionString))
            .applyToConnectionPoolSettings(b -> b.applyConnectionString(connectionString))
            .applyToServerSettings(b -> b.applyConnectionString(connectionString))
            .applyToSslSettings(b -> b.applyConnectionString(connectionString))
            .applyToSocketSettings(b -> b.applyConnectionString(connectionString))
            .codecRegistry(fromRegistries(
                MongoClients.getDefaultCodecRegistry(),
                fromProviders(PojoCodecProvider.builder()
                                               .automatic(true)
                                               .register(News.class)
                                               .build())
            ));

    if (connectionString.getReadPreference() != null) {
        builder.readPreference(connectionString.getReadPreference());
    }
    if (connectionString.getReadConcern() != null) {
        builder.readConcern(connectionString.getReadConcern());
    }
    if (connectionString.getWriteConcern() != null) {
        builder.writeConcern(connectionString.getWriteConcern());
    }
    if (connectionString.getApplicationName() != null) {
        builder.applicationName(connectionString.getApplicationName());
    }
    return MongoClients.create(builder.build());
}
 

default MongoClient mongoClient() {
    ConnectionString connectionString = new ConnectionString("mongodb://localhost/news");
    MongoClientSettings.Builder builder = MongoClientSettings.builder()
            .streamFactoryFactory(NettyStreamFactory::new)
            .applyToClusterSettings((cs) -> cs
                    .applyConnectionString(connectionString))
            .applyToConnectionPoolSettings(cps -> cps
                    .applyConnectionString(connectionString))
            .applyToServerSettings(ss -> ss
                    .applyConnectionString(connectionString))
          // TODO: Do not work with JDK11 without the next line being commented (null is not allowed)
          //.credential(connectionString.getCredential())
            .applyToSslSettings(ss -> ss
                    .applyConnectionString(connectionString))
            .applyToSocketSettings(ss -> ss
                    .applyConnectionString(connectionString))
            .codecRegistry(fromRegistries(
                    MongoClients.getDefaultCodecRegistry(),
                    fromProviders(PojoCodecProvider.builder()
                            .automatic(true)
                            .register(News.class)
                            .build())
            ));

    if (connectionString.getReadPreference() != null) {
        builder.readPreference(connectionString.getReadPreference());
    }
    if (connectionString.getReadConcern() != null) {
        builder.readConcern(connectionString.getReadConcern());
    }
    if (connectionString.getWriteConcern() != null) {
        builder.writeConcern(connectionString.getWriteConcern());
    }
    if (connectionString.getApplicationName() != null) {
        builder.applicationName(connectionString.getApplicationName());
    }

    return MongoClients.create(builder.build());
}
 

private MongoClientWrapper(final MongoClient theMongoClient,
        final String defaultDatabaseName,
        final DittoMongoClientSettings theDittoMongoClientSettings,
        @Nullable final EventLoopGroup theEventLoopGroup) {

    mongoClient = theMongoClient;
    defaultDatabase = theMongoClient.getDatabase(defaultDatabaseName);
    dittoMongoClientSettings = theDittoMongoClientSettings;
    eventLoopGroup = theEventLoopGroup;
}
 

@Bean
public MongoClient mongoClient() throws IOException {
  ServerAddress serverAddress = getServerAddress();
  MongoClientSettings settings = MongoClientSettings.builder()
                                                    .clusterSettings(ClusterSettings.builder()
                                                                                    .hosts(singletonList(serverAddress))
                                                                                    .requiredClusterType(STANDALONE)
                                                                                    .build()).build();
  return MongoClients.create(settings);
}
 

@Override
public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) throws ParameterResolutionException {
  final Class<?> type = parameterContext.getParameter().getType();
  if (MongoDatabase.class.isAssignableFrom(type)) {
    return getOrCreate(extensionContext).instance.database();
  } else if (MongoClient.class.isAssignableFrom(type)) {
    return getOrCreate(extensionContext).instance.client();
  }

  throw new ExtensionConfigurationException(String.format("%s supports only %s or %s but yours was %s", MongoExtension.class.getSimpleName(),
          MongoDatabase.class.getName(), MongoClient.class.getName(), type.getName()));
}
 

@Override
public MongoClient reactiveMongoClient() {
    return MongoClients.create();
}
 

public MongoEventStore(MongoClient mongoClient, IEventSerializer eventSerializer) {
    this(mongoClient, new MongoConfiguration(), eventSerializer);
}
 

public MongoPublishedVersionStore(MongoClient mongoClient) {
    this(mongoClient, new MongoConfiguration());
}
 

public MongoPublishedVersionStore(MongoClient mongoClient, MongoConfiguration configuration) {
    this.configuration = configuration;
    this.mongoClient = mongoClient;
    this.uniqueIndexName = configuration.getPublishedVersionUniqueIndexName();
    this.duplicateCode = configuration.getDuplicateCode();
}
 

@Bean
@ConditionalOnProperty(prefix = "spring.enode", name = "eventstore", havingValue = "mongo")
public MongoEventStore mongoEventStore(@Qualifier("enodeMongoClient") MongoClient mongoClient, IEventSerializer eventSerializer) {
    MongoEventStore eventStore = new MongoEventStore(mongoClient, eventSerializer);
    return eventStore;
}
 

@Bean
@ConditionalOnProperty(prefix = "spring.enode", name = "eventstore", havingValue = "mongo")
public MongoPublishedVersionStore mongoPublishedVersionStore(@Qualifier("enodeMongoClient") MongoClient mongoClient) {
    return new MongoPublishedVersionStore(mongoClient);
}
 

@Bean("enodeMongoClient")
@ConditionalOnProperty(prefix = "spring.enode", name = "eventstore", havingValue = "mongo")
public MongoClient mongoClient() {
    return MongoClients.create();
}
 

@Override
public MongoClient reactiveMongoClient() {
    return MongoClients.create();
}
 

public ReactiveMongoClientImpl(MongoClient client) {
    this.client = client;
}
 

@Override
public MongoClient unwrap() {
    return client;
}
 

@Override
public MongoClient mongoClient() {
	return MongoClients.create();
}
 

@Override
public MongoClient reactiveMongoClient() {
    return MongoClients.create(mongoUri);
}
 

@Override
public MongoClient reactiveMongoClient() {
    return MongoClients.create(mongoUri);
}
 

@Override
public MongoClient reactiveMongoClient() {
    return MongoClients.create(mongoUri);
}
 

@Override
public MongoClient reactiveMongoClient() {
    return MongoClients.create(mongoUri);
}
 

@Override
public MongoClient reactiveMongoClient() {
    return MongoClients.create(mongoUri);
}