|
20 | 20 |
|
21 | 21 | package com.arangodb.internal; |
22 | 22 |
|
23 | | -import java.io.IOException; |
24 | | -import java.lang.reflect.Type; |
25 | | -import java.util.concurrent.CompletableFuture; |
26 | | - |
27 | 23 | import com.arangodb.ArangoDBException; |
28 | 24 | import com.arangodb.internal.net.HostHandle; |
29 | 25 | import com.arangodb.internal.util.ArangoSerializationFactory; |
30 | 26 | import com.arangodb.internal.velocystream.VstCommunicationAsync; |
31 | 27 | import com.arangodb.velocypack.exception.VPackException; |
32 | 28 | import com.arangodb.velocystream.Request; |
33 | 29 |
|
| 30 | +import java.io.IOException; |
| 31 | +import java.lang.reflect.Type; |
| 32 | +import java.util.concurrent.CompletableFuture; |
| 33 | +import java.util.concurrent.ExecutorService; |
| 34 | +import java.util.concurrent.Executors; |
| 35 | + |
34 | 36 | /** |
35 | 37 | * @author Mark Vollmary |
36 | | - * |
| 38 | + * @author Michele Rastelli |
37 | 39 | */ |
38 | 40 | public class ArangoExecutorAsync extends ArangoExecutor { |
39 | 41 |
|
40 | | - private final VstCommunicationAsync communication; |
| 42 | + private final VstCommunicationAsync communication; |
| 43 | + private final ExecutorService outgoingExecutor = Executors.newSingleThreadExecutor(); |
| 44 | + |
| 45 | + public ArangoExecutorAsync(final VstCommunicationAsync communication, final ArangoSerializationFactory util, |
| 46 | + final DocumentCache documentCache) { |
| 47 | + super(util, documentCache); |
| 48 | + this.communication = communication; |
| 49 | + } |
41 | 50 |
|
42 | | - public ArangoExecutorAsync(final VstCommunicationAsync communication, final ArangoSerializationFactory util, |
43 | | - final DocumentCache documentCache) { |
44 | | - super(util, documentCache); |
45 | | - this.communication = communication; |
46 | | - } |
| 51 | + public <T> CompletableFuture<T> execute(final Request request, final Type type) { |
| 52 | + return execute(request, (response) -> createResult(type, response)); |
| 53 | + } |
47 | 54 |
|
48 | | - public <T> CompletableFuture<T> execute(final Request request, final Type type) { |
49 | | - return execute(request, (response) -> createResult(type, response)); |
50 | | - } |
| 55 | + public <T> CompletableFuture<T> execute(final Request request, final Type type, final HostHandle hostHandle) { |
| 56 | + return execute(request, (response) -> createResult(type, response), hostHandle); |
| 57 | + } |
51 | 58 |
|
52 | | - public <T> CompletableFuture<T> execute(final Request request, final Type type, final HostHandle hostHandle) { |
53 | | - return execute(request, (response) -> createResult(type, response), hostHandle); |
54 | | - } |
| 59 | + public <T> CompletableFuture<T> execute(final Request request, final ResponseDeserializer<T> responseDeserializer) { |
| 60 | + return execute(request, responseDeserializer, null); |
| 61 | + } |
55 | 62 |
|
56 | | - public <T> CompletableFuture<T> execute(final Request request, final ResponseDeserializer<T> responseDeserializer) { |
57 | | - return execute(request, responseDeserializer, null); |
58 | | - } |
| 63 | + public <T> CompletableFuture<T> execute( |
| 64 | + final Request request, |
| 65 | + final ResponseDeserializer<T> responseDeserializer, |
| 66 | + final HostHandle hostHandle) { |
59 | 67 |
|
60 | | - public <T> CompletableFuture<T> execute( |
61 | | - final Request request, |
62 | | - final ResponseDeserializer<T> responseDeserializer, |
63 | | - final HostHandle hostHandle) { |
64 | | - final CompletableFuture<T> result = new CompletableFuture<>(); |
65 | | - communication.execute(request, hostHandle).whenComplete((response, ex) -> { |
66 | | - if (response != null) { |
67 | | - try { |
68 | | - result.complete(responseDeserializer.deserialize(response)); |
69 | | - } catch (final VPackException | ArangoDBException e) { |
70 | | - result.completeExceptionally(e); |
71 | | - } |
72 | | - } else if (ex != null) { |
73 | | - result.completeExceptionally(ex); |
74 | | - } else { |
75 | | - result.cancel(true); |
76 | | - } |
77 | | - }); |
78 | | - return result; |
79 | | - } |
| 68 | + CompletableFuture<T> result = new CompletableFuture<>(); |
| 69 | + outgoingExecutor.execute(() -> { |
| 70 | + try { |
| 71 | + communication.execute(request, hostHandle) |
| 72 | + .whenCompleteAsync((response, ex) -> { |
| 73 | + if (ex != null) { |
| 74 | + result.completeExceptionally(ex); |
| 75 | + } else if (response != null) { |
| 76 | + try { |
| 77 | + result.complete(responseDeserializer.deserialize(response)); |
| 78 | + } catch (final VPackException | ArangoDBException e) { |
| 79 | + result.completeExceptionally(e); |
| 80 | + } |
| 81 | + } |
| 82 | + }); |
| 83 | + } catch (ArangoDBException e) { |
| 84 | + result.completeExceptionally(e); |
| 85 | + } |
| 86 | + } |
| 87 | + ); |
| 88 | + return result; |
| 89 | + } |
80 | 90 |
|
81 | | - public void disconnect() throws IOException { |
82 | | - communication.close(); |
83 | | - } |
| 91 | + public void disconnect() throws IOException { |
| 92 | + communication.close(); |
| 93 | + } |
84 | 94 | } |
0 commit comments