Java 类org.apache.camel.component.elasticsearch.ElasticsearchComponent 实例源码

@Bean
@ConditionalOnClass(CamelContext.class)
@ConditionalOnMissingBean(ElasticsearchComponent.class)
public ElasticsearchComponent configureElasticsearchComponent(
        CamelContext camelContext,
        ElasticsearchComponentConfiguration configuration) throws Exception {
    ElasticsearchComponent component = new ElasticsearchComponent();
    component.setCamelContext(camelContext);
    Map<String, Object> parameters = new HashMap<>();
    IntrospectionSupport.getProperties(configuration, parameters, null,
            false);
    IntrospectionSupport.setProperties(camelContext,
            camelContext.getTypeConverter(), component, parameters);
    return component;
}

@Test
public void testIndexContentUsingHeaders() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:start").to("elasticsearch://local");
        }
    });

    camelctx.getComponent("elasticsearch", ElasticsearchComponent.class).setClient(client);
    camelctx.start();
    try {
        Map<String, String> indexedData = new HashMap<>();
        indexedData.put("content", "test");

        Map<String, Object> headers = new HashMap<>();
        headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchConstants.OPERATION_INDEX);
        headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
        headers.put(ElasticsearchConstants.PARAM_INDEX_TYPE, "tweet");

        ProducerTemplate template = camelctx.createProducerTemplate();

        String indexId = template.requestBodyAndHeaders("direct:start", indexedData, headers, String.class);
        Assert.assertNotNull("Index id should not be null", indexId);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testGetContent() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:index").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet");
            from("direct:get").to("elasticsearch://local?operation=GET_BY_ID&indexName=twitter&indexType=tweet");
        }
    });

    camelctx.getComponent("elasticsearch", ElasticsearchComponent.class).setClient(client);
    camelctx.start();
    try {
        Map<String, String> indexedData = new HashMap<>();
        indexedData.put("content", "test");

        // Index some initial data
        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBody("direct:index", indexedData);

        String indexId = template.requestBody("direct:index", indexedData, String.class);
        Assert.assertNotNull("Index id should not be null", indexId);

        // Retrieve indexed data
        GetResponse response = template.requestBody("direct:get", indexId, GetResponse.class);
        Assert.assertNotNull("getResponse should not be null", response);
    } finally {
        camelctx.stop();
    }
}

@Test
public void testSearchContent() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:index").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet");
            from("direct:search").to("elasticsearch://local?operation=SEARCH&indexName=twitter&indexType=tweet");
        }
    });

    camelctx.getComponent("elasticsearch", ElasticsearchComponent.class).setClient(client);
    camelctx.start();
    try {
        Map<String, String> indexedData = new HashMap<>();
        indexedData.put("content", "test");

        // Index some initial data
        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBody("direct:index", indexedData);

        // Search for content
        Map<String, Object> actualQuery = new HashMap<>();
        actualQuery.put("content", "searchtest");
        Map<String, Object> match = new HashMap<>();
        match.put("match", actualQuery);
        Map<String, Object> query = new HashMap<>();
        query.put("query", match);

        SearchResponse searchResponse = template.requestBody("direct:search", query, SearchResponse.class);

        Assert.assertNotNull("searchResponse should not be null", searchResponse);
        Assert.assertNotNull("searchResponse hit count should equal 1", searchResponse.getHits().totalHits());
    } finally {
        camelctx.stop();
    }
}

public K8ElasticSearchEndpoint(String uri, ElasticsearchComponent component, K8ElasticSearchConfiguration config, Client client) throws Exception {
    super(uri, component,config,client);
    this.elasticSearchConfiguration = config;

}

@Test
public void testDeleteContent() throws Exception {

    CamelContext camelctx = new DefaultCamelContext();
    camelctx.addRoutes(new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            from("direct:index").to("elasticsearch://local?operation=INDEX&indexName=twitter&indexType=tweet");
            from("direct:get").to("elasticsearch://local?operation=GET_BY_ID&indexName=twitter&indexType=tweet");
            from("direct:delete").to("elasticsearch://local?operation=DELETE&indexName=twitter&indexType=tweet");
        }
    });

    camelctx.getComponent("elasticsearch", ElasticsearchComponent.class).setClient(client);
    camelctx.start();
    try {
        Map<String, String> indexedData = new HashMap<>();
        indexedData.put("content", "test");

        // Index some initial data
        ProducerTemplate template = camelctx.createProducerTemplate();
        template.sendBody("direct:index", indexedData);

        String indexId = template.requestBody("direct:index", indexedData, String.class);
        Assert.assertNotNull("Index id should not be null", indexId);

        // Retrieve indexed data
        GetResponse getResponse = template.requestBody("direct:get", indexId, GetResponse.class);
        Assert.assertNotNull("getResponse should not be null", getResponse);

        // Delete indexed data
        DeleteResponse deleteResponse = template.requestBody("direct:delete", indexId, DeleteResponse.class);
        Assert.assertNotNull("deleteResponse should not be null", deleteResponse);

        // Verify that the data has been deleted
        getResponse = template.requestBody("direct:get", indexId, GetResponse.class);
        Assert.assertNotNull("getResponse should not be null", getResponse);
        Assert.assertNull("getResponse source should be null", getResponse.getSource());
    } finally {
        camelctx.stop();
    }
}