Merge pull request #83 from eclipse/diana-driver-62

Connect to the ES(Elasticsearch) via http

Author: otaviojava

elasticsearch-driver/README.md

@@ -6,8 +6,14 @@
 
 ### How To Install
 
-1. Download the code: https://www.elastic.co/downloads/elasticsearch
-1. Follow the install and run steps: https://www.elastic.co/downloads/elasticsearch
+Once this a communication layer to Elasticsearch, we're using integration test, so you need to install Elasticsearch. The recommended way is using Docker.
+
+![Docker](https://www.docker.com/sites/default/files/horizontal_large.png)
+
+1. Install docker: https://www.docker.com/
+1. https://www.elastic.co/guide/en/elasticsearch/reference/current/docker.html
+1. Run docker command
+1. `docker run -p 9200:9200 -p 9300:9300 -e "discovery.type=single-node" docker.elastic.co/elasticsearch/elasticsearch:6.2.3`
 1. Execute the maven install `mvn clean install`
 
 

elasticsearch-driver/pom.xml

@@ -60,11 +60,12 @@
         <artifactId>diana-driver</artifactId>
         <version>0.0.5-SNAPSHOT</version>
     </parent>
+
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compile.targetLevel>1.8</maven.compile.targetLevel>
         <maven.compile.sourceLevel>1.8</maven.compile.sourceLevel>
-        <arango.driver>4.1.4</arango.driver>
+        <es.version>6.2.3</es.version>
     </properties>
     <dependencies>
         <dependency>
@@ -79,13 +80,8 @@
         </dependency>
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
-            <artifactId>transport</artifactId>
-            <version>5.6.3</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.logging.log4j</groupId>
-            <artifactId>log4j-to-slf4j</artifactId>
-            <version>2.9.1</version>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+            <version>${es.version}</version>
         </dependency>
     </dependencies>
     <build>

elasticsearch-driver/src/main/java/org/jnosql/diana/elasticsearch/document/DefaultElasticsearchDocumentCollectionManager.java

@@ -15,27 +15,28 @@
 package org.jnosql.diana.elasticsearch.document;
 
 
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.client.Client;
+import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.jnosql.diana.api.document.Document;
 import org.jnosql.diana.api.document.DocumentDeleteQuery;
 import org.jnosql.diana.api.document.DocumentEntity;
 import org.jnosql.diana.api.document.DocumentQuery;
 
-import javax.json.bind.Jsonb;
-import javax.json.bind.JsonbBuilder;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.StreamSupport.stream;
-import static org.elasticsearch.common.unit.TimeValue.timeValueMillis;
 import static org.jnosql.diana.elasticsearch.document.EntityConverter.ID_FIELD;
 import static org.jnosql.diana.elasticsearch.document.EntityConverter.getMap;
 
@@ -45,52 +46,35 @@
 class DefaultElasticsearchDocumentCollectionManager implements ElasticsearchDocumentCollectionManager {
 
 
-    protected static final Jsonb JSONB = JsonbBuilder.create();
-
-    private final Client client;
+    private final RestHighLevelClient client;
 
     private final String index;
 
-    DefaultElasticsearchDocumentCollectionManager(Client client, String index) {
+    DefaultElasticsearchDocumentCollectionManager(RestHighLevelClient client, String index) {
         this.client = client;
         this.index = index;
     }
 
     @Override
-    public DocumentEntity insert(DocumentEntity entity) throws NullPointerException {
+    public DocumentEntity insert(DocumentEntity entity) {
         requireNonNull(entity, "entity is required");
         Document id = entity.find(ID_FIELD)
                 .orElseThrow(() -> new ElasticsearchKeyFoundException(entity.toString()));
         Map<String, Object> jsonObject = getMap(entity);
-        byte[] bytes = JSONB.toJson(jsonObject).getBytes(UTF_8);
+        IndexRequest request = new IndexRequest(index, entity.getName(), id.get(String.class)).source(jsonObject);
         try {
-            client.prepareIndex(index, entity.getName(), id.get(String.class)).setSource(bytes)
-                    .execute().get();
-            return entity;
-        } catch (InterruptedException | ExecutionException e) {
-            throw new ElasticsearchException("An error to try to save/update entity on elasticsearch", e);
+            client.index(request);
+        } catch (IOException e) {
+            throw new ElasticsearchException("An error to insert in Elastic search", e);
         }
 
+        return entity;
     }
 
 
     @Override
-    public DocumentEntity insert(DocumentEntity entity, Duration ttl) throws NullPointerException, UnsupportedOperationException {
-        requireNonNull(entity, "entity is required");
-        requireNonNull(ttl, "ttl is required");
-        Document id = entity.find(ID_FIELD)
-                .orElseThrow(() -> new ElasticsearchKeyFoundException(entity.toString()));
-        Map<String, Object> jsonObject = getMap(entity);
-        byte[] bytes = JSONB.toJson(jsonObject).getBytes(UTF_8);
-        try {
-            client.prepareIndex(index, entity.getName(), id.get(String.class))
-                    .setSource(bytes)
-                    .setTTL(timeValueMillis(ttl.toMillis()))
-                    .execute().get();
-        } catch (InterruptedException | ExecutionException e) {
-            throw new ElasticsearchException("An error to try to save with TTL entity on elasticsearch", e);
-        }
-        return entity;
+    public DocumentEntity insert(DocumentEntity entity, Duration ttl) {
+        throw new UnsupportedOperationException("The insert with TTL does not support");
     }
 
     @Override
@@ -102,21 +86,27 @@ public DocumentEntity update(DocumentEntity entity) throws NullPointerException
     public void delete(DocumentDeleteQuery query) throws NullPointerException {
         requireNonNull(query, "query is required");
 
-       query.getCondition().orElseThrow(() -> new IllegalArgumentException("condition is required"));
-        DocumentQuery select  = new ElasticsearchDocumentQuery(query);
+        query.getCondition().orElseThrow(() -> new IllegalArgumentException("condition is required"));
+        DocumentQuery select = new ElasticsearchDocumentQuery(query);
 
         List<DocumentEntity> entities = select(select);
 
+        if (entities.isEmpty()) {
+            return;
+        }
+
+        BulkRequest bulk = new BulkRequest();
+
         entities.stream()
                 .map(entity -> entity.find(ID_FIELD).get().get(String.class))
-                .forEach(id -> {
-                    try {
-                        client.prepareDelete(index, query.getDocumentCollection(), id).execute().get();
-                    } catch (InterruptedException | ExecutionException e) {
-                        throw new ElasticsearchException("An error to delete entities on elasticsearch", e);
-                    }
-                });
+                .map(id -> new DeleteRequest(index, query.getDocumentCollection(), id))
+                .forEach(bulk::add);
 
+        try {
+            client.bulk(bulk);
+        } catch (IOException e) {
+            throw new ElasticsearchException("An error to delete entities on elasticsearch", e);
+        }
     }
 
 
@@ -130,19 +120,19 @@ public List<DocumentEntity> select(DocumentQuery query) throws NullPointerExcept
     public List<DocumentEntity> search(QueryBuilder query, String... types) throws NullPointerException {
         Objects.requireNonNull(query, "query is required");
 
-        SearchResponse searchResponse = null;
         try {
-            searchResponse = client.prepareSearch(index)
-                    .setTypes(types)
-                    .setQuery(query)
-                    .execute().get();
-
-            return stream(searchResponse.getHits().spliterator(), false)
-                    .map(h -> new ElasticsearchEntry(h.getId(), h.getIndex(), h.sourceAsMap()))
+            SearchRequest searchRequest = new SearchRequest(index);
+            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+            searchSourceBuilder.query(query);
+            searchRequest.types(types);
+            SearchResponse search = client.search(searchRequest);
+
+            return stream(search.getHits().spliterator(), false)
+                    .map(h -> new ElasticsearchEntry(h.getId(), h.getIndex(), h.getSourceAsMap()))
                     .filter(ElasticsearchEntry::isNotEmpty)
                     .map(ElasticsearchEntry::toEntity)
                     .collect(Collectors.toList());
-        } catch (InterruptedException | ExecutionException e) {
+        } catch (IOException e) {
             throw new ElasticsearchException("An error when do search from QueryBuilder on elasticsearch", e);
         }
 
@@ -151,6 +141,10 @@ public List<DocumentEntity> search(QueryBuilder query, String... types) throws N
 
     @Override
     public void close() {
-
+        try {
+            client.close();
+        } catch (IOException e) {
+            throw new ElasticsearchException("An error when close the client", e);
+        }
     }
 }