remove MapBinder guice binding for processors, use ProcessorsRegistry instead
This commit is contained in:
parent
94469d75f9
commit
f651f5a531
|
@ -20,10 +20,6 @@
|
|||
package org.elasticsearch.ingest;
|
||||
|
||||
import org.elasticsearch.common.inject.AbstractModule;
|
||||
import org.elasticsearch.common.inject.multibindings.MapBinder;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Registry for processor factories
|
||||
|
@ -32,20 +28,21 @@ import java.util.Map;
|
|||
*/
|
||||
public class ProcessorsModule extends AbstractModule {
|
||||
|
||||
private final Map<String, ProcessorFactoryProvider> processorFactoryProviders = new HashMap<>();
|
||||
private final ProcessorsRegistry processorsRegistry;
|
||||
|
||||
public ProcessorsModule() {
|
||||
this.processorsRegistry = new ProcessorsRegistry();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void configure() {
|
||||
MapBinder<String, ProcessorFactoryProvider> mapBinder = MapBinder.newMapBinder(binder(), String.class, ProcessorFactoryProvider.class);
|
||||
for (Map.Entry<String, ProcessorFactoryProvider> entry : processorFactoryProviders.entrySet()) {
|
||||
mapBinder.addBinding(entry.getKey()).toInstance(entry.getValue());
|
||||
}
|
||||
bind(ProcessorsRegistry.class).toInstance(processorsRegistry);
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds a processor factory under a specific type name.
|
||||
*/
|
||||
public void addProcessor(String type, ProcessorFactoryProvider processorFactoryProvider) {
|
||||
processorFactoryProviders.put(type, processorFactoryProvider);
|
||||
processorsRegistry.addProcessor(type, processorFactoryProvider);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -0,0 +1,44 @@
|
|||
/*
|
||||
* Licensed to Elasticsearch under one or more contributor
|
||||
* license agreements. See the NOTICE file distributed with
|
||||
* this work for additional information regarding copyright
|
||||
* ownership. Elasticsearch licenses this file to you under
|
||||
* the Apache License, Version 2.0 (the "License"); you may
|
||||
* not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.ingest;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
public class ProcessorsRegistry {
|
||||
|
||||
private final Map<String, ProcessorFactoryProvider> processorFactoryProviders = new HashMap<>();
|
||||
|
||||
/**
|
||||
* Adds a processor factory under a specific type name.
|
||||
*/
|
||||
public void addProcessor(String type, ProcessorFactoryProvider processorFactoryProvider) {
|
||||
processorFactoryProviders.put(type, processorFactoryProvider);
|
||||
}
|
||||
|
||||
public ProcessorFactoryProvider getProcessor(String type) {
|
||||
return processorFactoryProviders.get(type);
|
||||
}
|
||||
|
||||
public Set<Map.Entry<String, ProcessorFactoryProvider>> entrySet() {
|
||||
return processorFactoryProviders.entrySet();
|
||||
}
|
||||
}
|
|
@ -37,6 +37,7 @@ import org.elasticsearch.discovery.DiscoverySettings;
|
|||
import org.elasticsearch.env.Environment;
|
||||
import org.elasticsearch.gateway.GatewayService;
|
||||
import org.elasticsearch.ingest.ProcessorFactoryProvider;
|
||||
import org.elasticsearch.ingest.ProcessorsRegistry;
|
||||
import org.elasticsearch.script.ScriptService;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
|
@ -58,16 +59,16 @@ public class IngestBootstrapper extends AbstractLifecycleComponent implements Cl
|
|||
private final Environment environment;
|
||||
private final PipelineStore pipelineStore;
|
||||
private final PipelineExecutionService pipelineExecutionService;
|
||||
private final Map<String, ProcessorFactoryProvider> processorFactoryProvider;
|
||||
private final ProcessorsRegistry processorsRegistry;
|
||||
|
||||
@Inject
|
||||
public IngestBootstrapper(Settings settings, ThreadPool threadPool, Environment environment,
|
||||
ClusterService clusterService, TransportService transportService,
|
||||
Map<String, ProcessorFactoryProvider> processorFactoryProvider) {
|
||||
ProcessorsRegistry processorsRegistry) {
|
||||
super(settings);
|
||||
this.threadPool = threadPool;
|
||||
this.environment = environment;
|
||||
this.processorFactoryProvider = processorFactoryProvider;
|
||||
this.processorsRegistry = processorsRegistry;
|
||||
this.pipelineStore = new PipelineStore(settings, clusterService, transportService);
|
||||
this.pipelineExecutionService = new PipelineExecutionService(pipelineStore, threadPool);
|
||||
|
||||
|
@ -83,7 +84,7 @@ public class IngestBootstrapper extends AbstractLifecycleComponent implements Cl
|
|||
clusterService.add(this);
|
||||
this.pipelineStore = pipelineStore;
|
||||
this.pipelineExecutionService = pipelineExecutionService;
|
||||
this.processorFactoryProvider = null;
|
||||
this.processorsRegistry = null;
|
||||
}
|
||||
|
||||
public PipelineStore getPipelineStore() {
|
||||
|
@ -102,7 +103,7 @@ public class IngestBootstrapper extends AbstractLifecycleComponent implements Cl
|
|||
|
||||
@Inject
|
||||
public void setScriptService(ScriptService scriptService) {
|
||||
pipelineStore.buildProcessorFactoryRegistry(processorFactoryProvider, environment, scriptService);
|
||||
pipelineStore.buildProcessorFactoryRegistry(processorsRegistry, environment, scriptService);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
@ -42,6 +42,7 @@ import org.elasticsearch.env.Environment;
|
|||
import org.elasticsearch.index.IndexNotFoundException;
|
||||
import org.elasticsearch.ingest.Pipeline;
|
||||
import org.elasticsearch.ingest.ProcessorFactoryProvider;
|
||||
import org.elasticsearch.ingest.ProcessorsRegistry;
|
||||
import org.elasticsearch.ingest.TemplateService;
|
||||
import org.elasticsearch.ingest.Processor;
|
||||
import org.elasticsearch.plugin.ingest.transport.delete.DeletePipelineRequest;
|
||||
|
@ -85,10 +86,10 @@ public class PipelineStore extends AbstractComponent implements Closeable {
|
|||
this.client = client;
|
||||
}
|
||||
|
||||
public void buildProcessorFactoryRegistry(Map<String, ProcessorFactoryProvider> processorFactoryProviders, Environment environment, ScriptService scriptService) {
|
||||
public void buildProcessorFactoryRegistry(ProcessorsRegistry processorsRegistry, Environment environment, ScriptService scriptService) {
|
||||
Map<String, Processor.Factory> processorFactories = new HashMap<>();
|
||||
TemplateService templateService = new InternalTemplateService(scriptService);
|
||||
for (Map.Entry<String, ProcessorFactoryProvider> entry : processorFactoryProviders.entrySet()) {
|
||||
for (Map.Entry<String, ProcessorFactoryProvider> entry : processorsRegistry.entrySet()) {
|
||||
Processor.Factory processorFactory = entry.getValue().apply(environment, templateService);
|
||||
processorFactories.put(entry.getKey(), processorFactory);
|
||||
}
|
||||
|
|
|
@ -25,6 +25,7 @@ import org.elasticsearch.cluster.ClusterService;
|
|||
import org.elasticsearch.cluster.ClusterState;
|
||||
import org.elasticsearch.common.settings.Settings;
|
||||
import org.elasticsearch.env.Environment;
|
||||
import org.elasticsearch.ingest.ProcessorsRegistry;
|
||||
import org.elasticsearch.test.ESSingleNodeTestCase;
|
||||
import org.elasticsearch.threadpool.ThreadPool;
|
||||
import org.elasticsearch.transport.TransportService;
|
||||
|
@ -32,8 +33,6 @@ import org.hamcrest.Matchers;
|
|||
import org.junit.Before;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import java.util.Collections;
|
||||
|
||||
public class IngestTemplateTests extends ESSingleNodeTestCase {
|
||||
|
||||
private IngestBootstrapper bootstrapper;
|
||||
|
@ -51,7 +50,7 @@ public class IngestTemplateTests extends ESSingleNodeTestCase {
|
|||
ClusterService clusterService = Mockito.mock(ClusterService.class);
|
||||
TransportService transportService = Mockito.mock(TransportService.class);
|
||||
bootstrapper = new IngestBootstrapper(
|
||||
Settings.EMPTY, threadPool, environment, clusterService, transportService, Collections.emptyMap()
|
||||
Settings.EMPTY, threadPool, environment, clusterService, transportService, new ProcessorsRegistry()
|
||||
);
|
||||
bootstrapper.setClient(client());
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue