blob: 69be8d7dce7251f647857050278fedbbb9449f3f (
plain)
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
|
package bjc.utils.cli;
import static bjc.utils.cli.TerminalCodes.*;
import java.io.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;
/**
* Implementation of {@link Terminal} using {@link Reader} and {@link Writer}
*
* @author bjcul
*
*/
public class StreamTerminal implements Terminal, Runnable {
private SortedSet<Long> pendingRequests;
private ConcurrentMap<Long, String> pendingReplies;
private Lock replyLock;
private Condition replyCondition;
private Queue<String> pendingOutput;
private boolean running;
private Scanner inputScanner;
private Writer output;
private long currentRequest = -1;
/**
* Create a new stream terminal.
*
* @param input The input source
* @param output The output source
*/
public StreamTerminal(Reader input, Writer output) {
this.inputScanner = new Scanner(input);
this.output = output;
this.pendingRequests = new TreeSet<>();
this.pendingReplies = new ConcurrentHashMap<>();
this.pendingOutput = new ArrayDeque<>();
this.replyLock = new ReentrantLock();
this.replyCondition = replyLock.newCondition();
}
@Override
public void run() {
running = true;
try {
output.write(INFO_STARTCOMPROC.toString() + "\n");
} catch (IOException e) {
// TODO Consider if there is some better way to handle these
throw new RuntimeException(e);
}
overall: while (running && inputScanner.hasNextLine()) {
try {
while (!pendingOutput.isEmpty())
output.write(pendingOutput.remove());
String ln = inputScanner.nextLine();
String com = "";
int spcIdx = ln.indexOf(' ');
if (spcIdx == -1) {
com = ln;
} else {
com = ln.substring(0, spcIdx);
ln = ln.substring(spcIdx + 1);
}
comswt: switch (com) {
case "r": {
// General command format is 'r <request no.>,<reply>
String subRep = ln.substring(2);
// Process a reply
int comIndex = subRep.indexOf(',');
long repNo = 0;
if (comIndex == -1) {
// Reply to the oldest message by default
repNo = pendingRequests.first();
} else {
String repStr = subRep.substring(0, comIndex);
try {
repNo = Long.parseLong(repStr);
} catch (NumberFormatException nfex) {
output.write(ERROR_INVREPNO.toString() + "\n");
continue overall;
}
// Skip over the comma
subRep = subRep.substring(comIndex + 1);
}
if (!pendingRequests.contains(repNo)) {
output.write(ERROR_UNKREPNO.toString() + "\n");
continue overall;
}
pendingRequests.remove(repNo);
pendingReplies.put(repNo, subRep);
replyLock.lock();
replyCondition.signalAll();
replyLock.unlock();
break comswt;
}
case "q":
running = false;
break comswt;
default:
output.write(ERROR_UNRECCOM.toString() + "\n");
}
} catch (IOException ioex) {
throw new RuntimeException(ioex);
}
}
running = false;
try {
output.write(INFO_ENDCOMPROC.toString() + "\n");
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public long submitRequest(String req) {
long reqNo = currentRequest + 1;
currentRequest += 1;
pendingOutput.add(reqNo + " " + req + "\n");
pendingRequests.add(reqNo);
return reqNo;
}
@Override
public String awaitReply(long id) throws InterruptedException {
if (pendingReplies.containsKey(id))
return pendingReplies.get(id);
while (true) {
replyLock.lock();
replyCondition.await();
replyLock.unlock();
// Explanation: Since the reply map is add-only, the lock isn't actually
// protecting anything. We just want to wait until a response is received.
if (pendingReplies.containsKey(id))
return pendingReplies.get(id);
}
}
@Override
public Optional<String> checkReply(long id) {
return Optional.ofNullable(pendingReplies.get(id));
}
@Override
public String submitRequestSync(String req) throws InterruptedException {
return awaitReply(submitRequest(req));
}
// TODO add variants of the two blocking methods above with timeout support
}
|
