blob: 623ea543ffbf16f9c343142af3b299d70f971764 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.http.client.core.communication;
import com.yahoo.vespa.http.client.config.Endpoint;
import com.yahoo.vespa.http.client.core.Document;
import com.yahoo.vespa.http.client.core.ErrorCode;
import com.yahoo.vespa.http.client.core.OperationStatus;
import com.yahoo.vespa.http.client.core.ServerResponseException;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.Clock;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
/**
* Dummy implementation.
*
* @author dybis
*/
public class DryRunGatewayConnection implements GatewayConnection {
private final Endpoint endpoint;
private final Clock clock;
private Instant connectionTime = null;
private Instant lastPollTime = null;
/** Set to true to hold off responding with a result to any incoming operations until this is set false */
private boolean hold = false;
private final List<Document> held = new ArrayList<>();
/** If this is set, handshake operations will throw this exception */
private ServerResponseException throwThisOnHandshake = null;
/** If this is set, all write operations will throw this exception */
private IOException throwThisOnWrite = null;
public DryRunGatewayConnection(Endpoint endpoint, Clock clock) {
this.endpoint = endpoint;
this.clock = clock;
}
@Override
public InputStream write(List<Document> docs) throws IOException {
if (throwThisOnWrite != null)
throw throwThisOnWrite;
if (hold) {
held.addAll(docs);
return new ByteArrayInputStream("".getBytes(StandardCharsets.UTF_8));
}
else {
StringBuilder result = new StringBuilder();
for (Document doc : held)
result.append(okResponse(doc).render());
held.clear();
for (Document doc : docs)
result.append(okResponse(doc).render());
return new ByteArrayInputStream(result.toString().getBytes(StandardCharsets.UTF_8));
}
}
@Override
public InputStream poll() throws IOException {
lastPollTime = clock.instant();
return write(new ArrayList<>());
}
@Override
public Instant lastPollTime() { return lastPollTime; }
@Override
public InputStream drain() throws IOException {
return write(new ArrayList<>());
}
@Override
public boolean connect() {
connectionTime = clock.instant();
return true;
}
@Override
public Instant connectionTime() { return connectionTime; }
@Override
public Endpoint getEndpoint() {
return endpoint;
}
@Override
public void handshake() throws ServerResponseException {
if (throwThisOnHandshake != null)
throw throwThisOnHandshake;
}
@Override
public void close() { }
public void hold(boolean hold) {
this.hold = hold;
}
/** Returns the document currently held in this */
public List<Document> held() { return Collections.unmodifiableList(held); }
public void throwOnWrite(IOException throwThisOnWrite) {
this.throwThisOnWrite = throwThisOnWrite;
}
public void throwOnHandshake(ServerResponseException throwThisOnHandshake) {
this.throwThisOnHandshake = throwThisOnHandshake;
}
private OperationStatus okResponse(Document document) {
return new OperationStatus("ok", document.getOperationId(), ErrorCode.OK, false, "");
}
}
|
