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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.feedapi;
import com.yahoo.concurrent.SystemTimer;
import com.yahoo.config.subscription.ConfigSubscriber;
import com.yahoo.documentapi.messagebus.protocol.DocumentMessage;
import com.yahoo.documentapi.messagebus.protocol.DocumentProtocol;
import com.yahoo.feedhandler.InputStreamRequest;
import com.yahoo.messagebus.Message;
import com.yahoo.messagebus.routing.Route;
import com.yahoo.vespaclient.config.FeederConfig;
import java.util.logging.Level;
import java.util.logging.Logger;
/**
* Utility class for assigning properties to messages, either from implicit
* config values or from explicit values in requests.
*/
public class MessagePropertyProcessor implements ConfigSubscriber.SingleSubscriber<FeederConfig> {
private static final Logger log = Logger.getLogger(MessagePropertyProcessor.class.getName());
private static final boolean defaultCreateIfNonExistent = false;
private FeederOptions feederOptions = null;
private Route defaultRoute = null;
private long defaultTimeoutMillis = 0;
private boolean retryEnabled = true;
private boolean defaultAbortOnDocumentError = true;
private boolean defaultAbortOnSendError = true;
private boolean configChanged = false;
public MessagePropertyProcessor(FeederConfig config) {
configure(config);
}
public void setRoute(String routeOverride) {
defaultRoute = Route.parse(routeOverride);
}
public PropertySetter buildPropertySetter(InputStreamRequest request) {
String routeParam = null;
double timeoutParam = -1;
String priorityParam = null;
String abortOnDocErrorParam = null;
String abortOnFeedErrorParam = null;
String traceStr = null;
String createIfNonExistentParam = null;
Double totalTimeoutParam = null;
if (request != null) {
routeParam = request.getProperty("route");
String timeoutStr = request.getProperty("timeout");
if (timeoutStr != null) {
timeoutParam = Double.parseDouble(timeoutStr);
}
timeoutStr = request.getProperty("totaltimeout");
if (timeoutStr != null) {
totalTimeoutParam = Double.parseDouble(timeoutStr);
}
priorityParam = request.getProperty("priority");
traceStr = request.getProperty("tracelevel");
abortOnDocErrorParam = request.getProperty("abortondocumenterror");
abortOnFeedErrorParam = request.getProperty("abortonfeederror");
createIfNonExistentParam = request.getProperty("createifnonexistent");
}
Route route = (routeParam != null ? Route.parse(routeParam) : null);
long timeout;
boolean retry;
boolean abortOnDocumentError;
boolean abortOnFeedError;
boolean createIfNonExistent;
synchronized (this) {
if (route == null) {
route = defaultRoute;
}
timeout = (timeoutParam < 0 ? defaultTimeoutMillis : (long)(timeoutParam * 1000));
retry = retryEnabled;
abortOnDocumentError = (abortOnDocErrorParam == null ? defaultAbortOnDocumentError : (!"false".equals(abortOnDocErrorParam)));
abortOnFeedError = (abortOnFeedErrorParam == null ? defaultAbortOnSendError : (!"false".equals(abortOnFeedErrorParam)));
createIfNonExistent = (createIfNonExistentParam == null ? defaultCreateIfNonExistent : ("true".equals(createIfNonExistentParam)));
}
long totalTimeout = (totalTimeoutParam == null) ? timeout : (long)(totalTimeoutParam*1000);
DocumentProtocol.Priority priority = null;
if (priorityParam != null) {
priority = DocumentProtocol.getPriorityByName(priorityParam);
}
return new PropertySetter(route, timeout, totalTimeout, priority, retry, abortOnDocumentError, abortOnFeedError, createIfNonExistent, traceStr != null ? Integer.parseInt(traceStr) : 0);
}
public long getDefaultTimeoutMillis() { return defaultTimeoutMillis; }
synchronized boolean configChanged() {
return configChanged;
}
synchronized void setConfigChanged(boolean configChanged) {
this.configChanged = configChanged;
}
synchronized FeederOptions getFeederOptions() {
return feederOptions;
}
public synchronized void configure(FeederConfig config) {
if (feederOptions != null) {
setConfigChanged(true);
}
feederOptions = new FeederOptions(config);
if (feederOptions.getRoute() != null) {
defaultRoute = Route.parse(feederOptions.getRoute());
} else {
defaultRoute = null;
}
defaultTimeoutMillis = (long) (feederOptions.getTimeout() * 1000);
retryEnabled = feederOptions.getRetryEnabled();
defaultAbortOnDocumentError = feederOptions.abortOnDocumentError();
defaultAbortOnSendError = feederOptions.abortOnSendError();
if (log.isLoggable(Level.FINE)) {
log.log(Level.FINE, "Received new config (" +
"route: " + (defaultRoute != null ? defaultRoute : "<none>") +
", timeout: " + defaultTimeoutMillis + " ms, retry enabled: " + retryEnabled +
", abort on doc error: " + defaultAbortOnDocumentError +
", abort on feed error: " + defaultAbortOnSendError + ")");
}
}
public class PropertySetter implements MessageProcessor {
/** Route either set by configuration or by explicit request override. May be null */
private Route route;
/** Timeout (in milliseconds) */
private final long timeout;
private final long totalTimeout;
private final long startTime;
/** Explicit priority set. May be null */
private DocumentProtocol.Priority priority;
private final boolean retryEnabled;
private final boolean abortOnDocumentError;
private final boolean abortOnFeedError;
private final boolean createIfNonExistent;
private final int traceLevel;
PropertySetter(Route route, long timeout, long totalTimeout, DocumentProtocol.Priority priority,
boolean retryEnabled, boolean abortOnDocumentError, boolean abortOnFeedError,
boolean createIfNonExistent, int traceLevel) {
this.route = route;
this.timeout = timeout;
this.totalTimeout = totalTimeout;
this.priority = priority;
this.retryEnabled = retryEnabled;
this.abortOnDocumentError = abortOnDocumentError;
this.abortOnFeedError = abortOnFeedError;
this.createIfNonExistent = createIfNonExistent;
this.traceLevel = traceLevel;
this.startTime = SystemTimer.INSTANCE.milliTime();
}
private long getTimeRemaining() {
return (totalTimeout < 0L)
? timeout
: Math.min(timeout, totalTimeout - (SystemTimer.INSTANCE.milliTime() - startTime));
}
public Route getRoute() {
return route;
}
public void setRoute(Route route) {
this.route = route;
}
public DocumentProtocol.Priority getPriority() {
return priority;
}
public void setPriority(DocumentProtocol.Priority priority) {
this.priority = priority;
}
public boolean getAbortOnDocumentError() {
return abortOnDocumentError;
}
public boolean getAbortOnFeedError() {
return abortOnFeedError;
}
public boolean getCreateIfNonExistent() {
return createIfNonExistent;
}
@Override
@SuppressWarnings("removal") // TODO Vespa 9: remove
public void process(Message msg) {
if (route != null) {
msg.setRoute(route);
}
msg.setTimeRemaining(getTimeRemaining());
msg.setRetryEnabled(retryEnabled);
msg.getTrace().setLevel(Math.max(getFeederOptions().getTraceLevel(), traceLevel));
if (priority != null) {
((DocumentMessage) msg).setPriority(priority);
}
}
}
}
|
