|
1 | 1 | package org.iot.dsa.dslink.requester;
|
2 | 2 |
|
3 |
| -import org.iot.dsa.dslink.DSRequestException; |
| 3 | +import java.util.ArrayList; |
4 | 4 | import org.iot.dsa.node.DSList;
|
5 | 5 | import org.iot.dsa.node.DSMap;
|
6 | 6 |
|
7 | 7 | /**
|
8 | 8 | * Action handler for non-tables/streams.
|
9 | 9 | * <p>
|
10 |
| - * Call getResult(long timeout) to block until the invocation is complete. It will either return |
| 10 | + * Call getUpdate(long timeout) to block until the invocation is complete. It will either return |
11 | 11 | * the result (possibly null), or throw an exception.
|
12 | 12 | *
|
13 | 13 | * @author Aaron Hansen
|
14 | 14 | */
|
15 | 15 | public class SimpleInvokeHandler extends AbstractInvokeHandler {
|
16 | 16 |
|
17 | 17 | ///////////////////////////////////////////////////////////////////////////
|
18 |
| - // Fields |
| 18 | + // Instance Fields |
19 | 19 | ///////////////////////////////////////////////////////////////////////////
|
20 | 20 |
|
21 |
| - private boolean autoClose = true; |
22 | 21 | private boolean closed = false;
|
| 22 | + private DSList columns; |
23 | 23 | private RuntimeException error;
|
24 |
| - private DSList result; |
| 24 | + private Mode mode; |
| 25 | + private DSMap tableMeta; |
| 26 | + private ArrayList<DSList> updates; |
25 | 27 |
|
26 | 28 | ///////////////////////////////////////////////////////////////////////////
|
27 |
| - // Methods |
| 29 | + // Public Methods |
28 | 30 | ///////////////////////////////////////////////////////////////////////////
|
29 | 31 |
|
30 | 32 | /**
|
31 |
| - * Waits for the stream to close before returning, or the timeout to occur. |
| 33 | + * It is possible for a values action to not return columns. |
32 | 34 | *
|
33 |
| - * @param timeout Passed to Object.wait |
34 |
| - * @return Null, or the first update. |
| 35 | + * @return Beware of null. |
| 36 | + */ |
| 37 | + public int getColumnCount() { |
| 38 | + if (columns == null) { |
| 39 | + return 0; |
| 40 | + } |
| 41 | + return columns.size(); |
| 42 | + } |
| 43 | + |
| 44 | + /** |
| 45 | + * The metadata map for the given column. |
| 46 | + */ |
| 47 | + public DSMap getColumnMetadata(int idx) { |
| 48 | + return columns.getMap(idx); |
| 49 | + } |
| 50 | + |
| 51 | + public RuntimeException getError() { |
| 52 | + return error; |
| 53 | + } |
| 54 | + |
| 55 | + public Mode getMode() { |
| 56 | + return mode; |
| 57 | + } |
| 58 | + |
| 59 | + public DSMap getTableMeta() { |
| 60 | + return tableMeta; |
| 61 | + } |
| 62 | + |
| 63 | + /** |
| 64 | + * The next available update, or null for actions return void. |
| 65 | + * Will wait for an update if one isn't available. Will return all updates before |
| 66 | + * throwing any exceptions. |
| 67 | + * |
| 68 | + * @param timeout How long to wait for an update or the stream to close. |
| 69 | + * @return Null if the action doesn't return anything. |
35 | 70 | * @throws RuntimeException if there is a timeout, or if there are any errors.
|
36 | 71 | */
|
37 |
| - public DSList getResult(long timeout) { |
| 72 | + public DSList getUpdate(long timeout) { |
| 73 | + long end = System.currentTimeMillis() + timeout; |
38 | 74 | synchronized (this) {
|
39 |
| - if (!closed) { |
| 75 | + while (!closed && !hasError() && !hasUpdates()) { |
40 | 76 | try {
|
41 | 77 | wait(timeout);
|
42 |
| - } catch (Exception x) { |
| 78 | + } catch (Exception expected) { |
| 79 | + } |
| 80 | + if (System.currentTimeMillis() > end) { |
| 81 | + break; |
43 | 82 | }
|
44 | 83 | }
|
45 |
| - } |
46 |
| - if (error != null) { |
47 |
| - throw error; |
48 |
| - } |
49 |
| - if (!closed) { |
| 84 | + if (hasUpdates()) { |
| 85 | + return updates.remove(0); |
| 86 | + } |
| 87 | + if (hasError()) { |
| 88 | + throw error; |
| 89 | + } |
| 90 | + if (closed) { |
| 91 | + return null; |
| 92 | + } |
50 | 93 | throw new IllegalStateException("Action timed out");
|
51 | 94 | }
|
52 |
| - return result; |
53 | 95 | }
|
54 | 96 |
|
55 | 97 | /**
|
56 |
| - * True by default, whether or not to close the stream upon receiving the first result. |
| 98 | + * Takes the updates such that subsequent calls will never return the same updates, except when |
| 99 | + * there are no updates in which case this returns null. |
| 100 | + * |
| 101 | + * @return Possibly null. |
57 | 102 | */
|
58 |
| - public boolean isAutoClose() { |
59 |
| - return autoClose; |
| 103 | + public synchronized ArrayList<DSList> getUpdates() { |
| 104 | + ArrayList<DSList> ret = updates; |
| 105 | + updates = null; |
| 106 | + return ret; |
60 | 107 | }
|
61 | 108 |
|
62 |
| - /** |
63 |
| - * Causes getResult to return. |
64 |
| - */ |
65 |
| - public void onClose() { |
66 |
| - synchronized (this) { |
67 |
| - closed = true; |
68 |
| - notifyAll(); |
69 |
| - } |
| 109 | + public boolean hasError() { |
| 110 | + return error != null; |
70 | 111 | }
|
71 | 112 |
|
72 |
| - /** |
73 |
| - * Will create an exception to be thrown by getResult. |
74 |
| - */ |
75 |
| - public void onError(ErrorType type, String msg) { |
76 |
| - synchronized (this) { |
77 |
| - error = ErrorType.makeException(type, msg); |
78 |
| - getStream().closeStream(); |
79 |
| - notifyAll(); |
| 113 | + public synchronized boolean hasUpdates() { |
| 114 | + if (updates == null) { |
| 115 | + return false; |
80 | 116 | }
|
| 117 | + return !updates.isEmpty(); |
81 | 118 | }
|
82 | 119 |
|
83 |
| - /** |
84 |
| - * Does nothing. |
85 |
| - */ |
86 |
| - public void onColumns(DSList list) { |
| 120 | + public boolean isClosed() { |
| 121 | + return closed; |
| 122 | + } |
| 123 | + |
| 124 | + @Override |
| 125 | + public synchronized void onClose() { |
| 126 | + closed = true; |
| 127 | + notifyAll(); |
| 128 | + } |
| 129 | + |
| 130 | + @Override |
| 131 | + public synchronized void onColumns(DSList list) { |
| 132 | + this.columns = list; |
| 133 | + notifyAll(); |
87 | 134 | }
|
88 | 135 |
|
89 | 136 | /**
|
90 |
| - * Will result in an error since tables and streams are not supported. |
| 137 | + * Creates an exception that will be thrown by getUpdate. |
91 | 138 | */
|
92 |
| - public void onInsert(int index, DSList rows) { |
93 |
| - synchronized (this) { |
94 |
| - error = new DSRequestException("Tables and streams not supported"); |
95 |
| - getStream().closeStream(); |
96 |
| - notifyAll(); |
97 |
| - } |
| 139 | + @Override |
| 140 | + public synchronized void onError(ErrorType type, String msg) { |
| 141 | + error = ErrorType.makeException(type, msg); |
| 142 | + notifyAll(); |
98 | 143 | }
|
99 | 144 |
|
100 | 145 | /**
|
101 |
| - * Does nothing. |
| 146 | + * Does nothing other than notify and threads waiting on this instance. |
102 | 147 | */
|
103 |
| - public void onMode(Mode mode) { |
| 148 | + @Override |
| 149 | + public synchronized void onInsert(int index, DSList rows) { |
| 150 | + notifyAll(); |
| 151 | + } |
| 152 | + |
| 153 | + @Override |
| 154 | + public synchronized void onMode(Mode mode) { |
| 155 | + this.mode = mode; |
| 156 | + notifyAll(); |
104 | 157 | }
|
105 | 158 |
|
106 | 159 | /**
|
107 |
| - * Will result in an error since tables and streams are not supported. |
| 160 | + * Does nothing other than notify and threads waiting on this instance. |
108 | 161 | */
|
109 |
| - public void onReplace(int start, int end, DSList rows) { |
110 |
| - synchronized (this) { |
111 |
| - error = new DSRequestException("Tables and streams not supported"); |
112 |
| - getStream().closeStream(); |
113 |
| - notifyAll(); |
114 |
| - } |
| 162 | + @Override |
| 163 | + public synchronized void onReplace(int start, int end, DSList rows) { |
| 164 | + notifyAll(); |
115 | 165 | }
|
116 | 166 |
|
117 |
| - public void onTableMeta(DSMap map) { |
| 167 | + @Override |
| 168 | + public synchronized void onTableMeta(DSMap map) { |
| 169 | + this.tableMeta = map; |
| 170 | + notifyAll(); |
118 | 171 | }
|
119 | 172 |
|
120 | 173 | /**
|
121 | 174 | * Captures the result and if auto-close is true, closes the stream.
|
122 | 175 | */
|
123 |
| - public void onUpdate(DSList row) { |
| 176 | + @Override |
| 177 | + public synchronized void onUpdate(DSList row) { |
| 178 | + if (updates == null) { |
| 179 | + updates = new ArrayList<>(); |
| 180 | + } |
| 181 | + updates.add(row); |
| 182 | + notifyAll(); |
| 183 | + } |
| 184 | + |
| 185 | + /** |
| 186 | + * Waits for any callback from the responder. Will return immediately if already closed. |
| 187 | + * |
| 188 | + * @param timeout Passed to Object.wait |
| 189 | + * @throws RuntimeException if there is an error with the invocation. |
| 190 | + * @throws IllegalStateException if there is a timeout, or if there are any errors. |
| 191 | + */ |
| 192 | + public void waitForCallback(long timeout) { |
124 | 193 | synchronized (this) {
|
125 |
| - result = row; |
126 |
| - if (autoClose) { |
127 |
| - getStream().closeStream(); |
| 194 | + if (!closed) { |
| 195 | + long end = System.currentTimeMillis() + timeout; |
| 196 | + try { |
| 197 | + wait(timeout); |
| 198 | + } catch (Exception x) { |
| 199 | + } |
| 200 | + if (System.currentTimeMillis() > end) { |
| 201 | + throw new IllegalStateException("Action timed out"); |
| 202 | + } |
128 | 203 | }
|
129 | 204 | }
|
130 | 205 | }
|
131 | 206 |
|
132 | 207 | /**
|
133 |
| - * Whether or not to auto close the stream on the first update. True by default, this |
134 |
| - * only needs to be called to disable. |
| 208 | + * Waits for the stream to close or the timeout to occur. |
| 209 | + * |
| 210 | + * @param timeout Passed to Object.wait |
| 211 | + * @throws IllegalStateException if there is a timeout, or if there are any errors. |
135 | 212 | */
|
136 |
| - public SimpleInvokeHandler setAutoClose(boolean arg) { |
137 |
| - autoClose = arg; |
138 |
| - return this; |
| 213 | + public void waitForClose(long timeout) { |
| 214 | + long end = System.currentTimeMillis() + timeout; |
| 215 | + synchronized (this) { |
| 216 | + while (!closed) { |
| 217 | + try { |
| 218 | + wait(timeout); |
| 219 | + } catch (Exception x) { |
| 220 | + } |
| 221 | + if (System.currentTimeMillis() > end) { |
| 222 | + break; |
| 223 | + } |
| 224 | + } |
| 225 | + } |
| 226 | + if (!closed) { |
| 227 | + throw new IllegalStateException("Action timed out"); |
| 228 | + } |
139 | 229 | }
|
140 | 230 |
|
141 | 231 | }
|
0 commit comments