File D:\maven-qcr-plugin-testsunitaires\qcr4java\h2-1.3.170-sources\target\site\qc\lecode\server\TcpServerThread.java.html

1 /*
2 * Copyright 2004-2011 H2 Group. Multiple-Licensed under the H2 License,
3 * Version 1.0, and under the Eclipse Public License, Version 1.0
4 * (http://h2database.com/html/license.html).
5 * Initial Developer: H2 Group
6 */
7 package org.h2.server;
8
9 import java.io.ByteArrayInputStream;
10 import java.io.FilterInputStream;
11 import java.io.IOException;
12 import java.io.InputStream;
13 import java.io.PrintWriter;
14 import java.io.StringWriter;
15 import java.net.Socket;
16 import java.sql.SQLException;
17 import java.util.ArrayList;
18 import org.h2.command.Command;
19 import org.h2.constant.ErrorCode;
20 import org.h2.constant.SysProperties;
21 import org.h2.engine.ConnectionInfo;
22 import org.h2.engine.Constants;
23 import org.h2.engine.Engine;
24 import org.h2.engine.Session;
25 import org.h2.engine.SessionRemote;
26 import org.h2.expression.Parameter;
27 import org.h2.expression.ParameterInterface;
28 import org.h2.expression.ParameterRemote;
29 import org.h2.jdbc.JdbcSQLException;
30 import org.h2.message.DbException;
31 import org.h2.result.ResultColumn;
32 import org.h2.result.ResultInterface;
33 import org.h2.store.LobStorage;
34 import org.h2.util.IOUtils;
35 import org.h2.util.SmallLRUCache;
36 import org.h2.util.SmallMap;
37 import org.h2.util.StringUtils;
38 import org.h2.value.Transfer;
39 import org.h2.value.Value;
40 import org.h2.value.ValueLobDb;
41
42 /**
43 * One server thread is opened per client connection.
44 */
45 public class TcpServerThread implements Runnable {
46
47 protected final Transfer transfer;
48 private final TcpServer server;
49 private Session session;
50 private boolean stop;
51 private Thread thread;
52 private Command commit;
53 private final SmallMap cache = new SmallMap(SysProperties.SERVER_CACHED_OBJECTS);
54 private final SmallLRUCache<Long, CachedInputStream> lobs =
55 SmallLRUCache.newInstance(Math.max(
56 SysProperties.SERVER_CACHED_OBJECTS,
57 SysProperties.SERVER_RESULT_SET_FETCH_SIZE * 5));
58 private final int threadId;
59 private int clientVersion;
60 private String sessionId;
61
62 TcpServerThread(Socket socket, TcpServer server, int id) {
63 this.server = server;
64 this.threadId = id;
65 transfer = new Transfer(null);
66 transfer.setSocket(socket);
67 }
68
69 private void trace(String s) {
70 server.trace(this + " " + s);
71 }
72
73 public void run() {
74 try {
75 transfer.init();
76 trace("Connect");
77 // TODO server: should support a list of allowed databases
78 // and a list of allowed clients
79 try {
80 if (!server.allow(transfer.getSocket())) {
81 throw DbException.get(ErrorCode.REMOTE_CONNECTION_NOT_ALLOWED);
82 }
83 int minClientVersion = transfer.readInt();
84 if (minClientVersion < Constants.TCP_PROTOCOL_VERSION_6) {
85 throw DbException.get(ErrorCode.DRIVER_VERSION_ERROR_2, "" + clientVersion, "" + Constants.TCP_PROTOCOL_VERSION_6);
86 } else if (minClientVersion > Constants.TCP_PROTOCOL_VERSION_12) {
87 throw DbException.get(ErrorCode.DRIVER_VERSION_ERROR_2, "" + clientVersion, "" + Constants.TCP_PROTOCOL_VERSION_12);
88 }
89 int maxClientVersion = transfer.readInt();
90 if (maxClientVersion >= Constants.TCP_PROTOCOL_VERSION_12) {
91 clientVersion = Constants.TCP_PROTOCOL_VERSION_12;
92 } else {
93 clientVersion = minClientVersion;
94 }
95 transfer.setVersion(clientVersion);
96 String db = transfer.readString();
97 String originalURL = transfer.readString();
98 if (db == null && originalURL == null) {
99 String targetSessionId = transfer.readString();
100 int command = transfer.readInt();
101 stop = true;
102 if (command == SessionRemote.SESSION_CANCEL_STATEMENT) {
103 // cancel a running statement
104 int statementId = transfer.readInt();
105 server.cancelStatement(targetSessionId, statementId);
106 } else if (command == SessionRemote.SESSION_CHECK_KEY) {
107 // check if this is the correct server
108 db = server.checkKeyAndGetDatabaseName(targetSessionId);
109 if (!targetSessionId.equals(db)) {
110 transfer.writeInt(SessionRemote.STATUS_OK);
111 } else {
112 transfer.writeInt(SessionRemote.STATUS_ERROR);
113 }
114 }
115 }
116 String baseDir = server.getBaseDir();
117 if (baseDir == null) {
118 baseDir = SysProperties.getBaseDir();
119 }
120 db = server.checkKeyAndGetDatabaseName(db);
121 ConnectionInfo ci = new ConnectionInfo(db);
122 if (baseDir != null) {
123 ci.setBaseDir(baseDir);
124 }
125 if (server.getIfExists()) {
126 ci.setProperty("IFEXISTS", "TRUE");
127 }
128 ci.setOriginalURL(originalURL);
129 ci.setUserName(transfer.readString());
130 ci.setUserPasswordHash(transfer.readBytes());
131 ci.setFilePasswordHash(transfer.readBytes());
132 int len = transfer.readInt();
133 for (int i = 0; i < len; i++) {
134 ci.setProperty(transfer.readString(), transfer.readString());
135 }
136 session = Engine.getInstance().createSession(ci);
137 transfer.setSession(session);
138 transfer.writeInt(SessionRemote.STATUS_OK);
139 transfer.writeInt(clientVersion);
140 transfer.flush();
141 server.addConnection(threadId, originalURL, ci.getUserName());
142 trace("Connected");
143 } catch (Throwable e) {
144 sendError(e);
145 stop = true;
146 }
147 while (!stop) {
148 try {
149 process();
150 } catch (Throwable e) {
151 sendError(e);
152 }
153 }
154 trace("Disconnect");
155 } catch (Throwable e) {
156 server.traceError(e);
157 } finally {
158 close();
159 }
160 }
161
162 private void closeSession() {
163 if (session != null) {
164 RuntimeException closeError = null;
165 try {
166 Command rollback = session.prepareLocal("ROLLBACK");
167 rollback.executeUpdate();
168 } catch (RuntimeException e) {
169 closeError = e;
170 server.traceError(e);
171 } catch (Exception e) {
172 server.traceError(e);
173 }
174 try {
175 session.close();
176 server.removeConnection(threadId);
177 } catch (RuntimeException e) {
178 if (closeError == null) {
179 closeError = e;
180 server.traceError(e);
181 }
182 } catch (Exception e) {
183 server.traceError(e);
184 } finally {
185 session = null;
186 }
187 if (closeError != null) {
188 throw closeError;
189 }
190 }
191 }
192
193 /**
194 * Close a connection.
195 */
196 void close() {
197 try {
198 stop = true;
199 closeSession();
200 } catch (Exception e) {
201 server.traceError(e);
202 } finally {
203 transfer.close();
204 trace("Close");
205 server.remove(this);
206 }
207 }
208
209 private void sendError(Throwable t) {
210 try {
211 SQLException e = DbException.convert(t).getSQLException();
212 StringWriter writer = new StringWriter();
213 e.printStackTrace(new PrintWriter(writer));
214 String trace = writer.toString();
215 String message;
216 String sql;
217 if (e instanceof JdbcSQLException) {
218 JdbcSQLException j = (JdbcSQLException) e;
219 message = j.getOriginalMessage();
220 sql = j.getSQL();
221 } else {
222 message = e.getMessage();
223 sql = null;
224 }
225 transfer.writeInt(SessionRemote.STATUS_ERROR).writeString(e.getSQLState()).writeString(message)
226 .writeString(sql).writeInt(e.getErrorCode()).writeString(trace).flush();
227 } catch (Exception e2) {
228 if (!transfer.isClosed()) {
229 server.traceError(e2);
230 }
231 // if writing the error does not work, close the connection
232 stop = true;
233 }
234 }
235
236 private void setParameters(Command command) throws IOException {
237 int len = transfer.readInt();
238 ArrayList<? extends ParameterInterface> params = command.getParameters();
239 for (int i = 0; i < len; i++) {
240 Parameter p = (Parameter) params.get(i);
241 p.setValue(transfer.readValue());
242 }
243 }
244
245 private void process() throws IOException {
246 int operation = transfer.readInt();
247 switch (operation) {
248 case SessionRemote.SESSION_PREPARE_READ_PARAMS:
249 case SessionRemote.SESSION_PREPARE: {
250 int id = transfer.readInt();
251 String sql = transfer.readString();
252 int old = session.getModificationId();
253 Command command = session.prepareLocal(sql);
254 boolean readonly = command.isReadOnly();
255 cache.addObject(id, command);
256 boolean isQuery = command.isQuery();
257 ArrayList<? extends ParameterInterface> params = command.getParameters();
258 transfer.writeInt(getState(old)).writeBoolean(isQuery).writeBoolean(readonly)
259 .writeInt(params.size());
260 if (operation == SessionRemote.SESSION_PREPARE_READ_PARAMS) {
261 for (ParameterInterface p : params) {
262 ParameterRemote.writeMetaData(transfer, p);
263 }
264 }
265 transfer.flush();
266 break;
267 }
268 case SessionRemote.SESSION_CLOSE: {
269 stop = true;
270 closeSession();
271 transfer.writeInt(SessionRemote.STATUS_OK).flush();
272 close();
273 break;
274 }
275 case SessionRemote.COMMAND_COMMIT: {
276 if (commit == null) {
277 commit = session.prepareLocal("COMMIT");
278 }
279 int old = session.getModificationId();
280 commit.executeUpdate();
281 transfer.writeInt(getState(old)).flush();
282 break;
283 }
284 case SessionRemote.COMMAND_GET_META_DATA: {
285 int id = transfer.readInt();
286 int objectId = transfer.readInt();
287 Command command = (Command) cache.getObject(id, false);
288 ResultInterface result = command.getMetaData();
289 cache.addObject(objectId, result);
290 int columnCount = result.getVisibleColumnCount();
291 transfer.writeInt(SessionRemote.STATUS_OK).writeInt(columnCount).writeInt(0);
292 for (int i = 0; i < columnCount; i++) {
293 ResultColumn.writeColumn(transfer, result, i);
294 }
295 transfer.flush();
296 break;
297 }
298 case SessionRemote.COMMAND_EXECUTE_QUERY: {
299 int id = transfer.readInt();
300 int objectId = transfer.readInt();
301 int maxRows = transfer.readInt();
302 int fetchSize = transfer.readInt();
303 Command command = (Command) cache.getObject(id, false);
304 setParameters(command);
305 int old = session.getModificationId();
306 ResultInterface result;
307 synchronized (session) {
308 result = command.executeQuery(maxRows, false);
309 }
310 cache.addObject(objectId, result);
311 int columnCount = result.getVisibleColumnCount();
312 int state = getState(old);
313 transfer.writeInt(state).writeInt(columnCount);
314 int rowCount = result.getRowCount();
315 transfer.writeInt(rowCount);
316 for (int i = 0; i < columnCount; i++) {
317 ResultColumn.writeColumn(transfer, result, i);
318 }
319 int fetch = Math.min(rowCount, fetchSize);
320 for (int i = 0; i < fetch; i++) {
321 sendRow(result);
322 }
323 transfer.flush();
324 break;
325 }
326 case SessionRemote.COMMAND_EXECUTE_UPDATE: {
327 int id = transfer.readInt();
328 Command command = (Command) cache.getObject(id, false);
329 setParameters(command);
330 int old = session.getModificationId();
331 int updateCount;
332 synchronized (session) {
333 updateCount = command.executeUpdate();
334 }
335 int status;
336 if (session.isClosed()) {
337 status = SessionRemote.STATUS_CLOSED;
338 } else {
339 status = getState(old);
340 }
341 transfer.writeInt(status).writeInt(updateCount).writeBoolean(session.getAutoCommit());
342 transfer.flush();
343 break;
344 }
345 case SessionRemote.COMMAND_CLOSE: {
346 int id = transfer.readInt();
347 Command command = (Command) cache.getObject(id, true);
348 if (command != null) {
349 command.close();
350 cache.freeObject(id);
351 }
352 break;
353 }
354 case SessionRemote.RESULT_FETCH_ROWS: {
355 int id = transfer.readInt();
356 int count = transfer.readInt();
357 ResultInterface result = (ResultInterface) cache.getObject(id, false);
358 transfer.writeInt(SessionRemote.STATUS_OK);
359 for (int i = 0; i < count; i++) {
360 sendRow(result);
361 }
362 transfer.flush();
363 break;
364 }
365 case SessionRemote.RESULT_RESET: {
366 int id = transfer.readInt();
367 ResultInterface result = (ResultInterface) cache.getObject(id, false);
368 result.reset();
369 break;
370 }
371 case SessionRemote.RESULT_CLOSE: {
372 int id = transfer.readInt();
373 ResultInterface result = (ResultInterface) cache.getObject(id, true);
374 if (result != null) {
375 result.close();
376 cache.freeObject(id);
377 }
378 break;
379 }
380 case SessionRemote.CHANGE_ID: {
381 int oldId = transfer.readInt();
382 int newId = transfer.readInt();
383 Object obj = cache.getObject(oldId, false);
384 cache.freeObject(oldId);
385 cache.addObject(newId, obj);
386 break;
387 }
388 case SessionRemote.SESSION_SET_ID: {
389 sessionId = transfer.readString();
390 transfer.writeInt(SessionRemote.STATUS_OK).flush();
391 break;
392 }
393 case SessionRemote.SESSION_SET_AUTOCOMMIT: {
394 boolean autoCommit = transfer.readBoolean();
395 session.setAutoCommit(autoCommit);
396 transfer.writeInt(SessionRemote.STATUS_OK).flush();
397 break;
398 }
399 case SessionRemote.SESSION_UNDO_LOG_POS: {
400 transfer.writeInt(SessionRemote.STATUS_OK).
401 writeInt(session.getUndoLogPos()).flush();
402 break;
403 }
404 case SessionRemote.LOB_READ: {
405 long lobId = transfer.readLong();
406 byte[] hmac;
407 CachedInputStream in;
408 if (clientVersion >= Constants.TCP_PROTOCOL_VERSION_11) {
409 if (clientVersion >= Constants.TCP_PROTOCOL_VERSION_12) {
410 hmac = transfer.readBytes();
411 transfer.verifyLobMac(hmac, lobId);
412 } else {
413 hmac = null;
414 }
415 in = lobs.get(lobId);
416 if (in == null) {
417 in = new CachedInputStream(null);
418 lobs.put(lobId, in);
419 }
420 } else {
421 hmac = null;
422 in = lobs.get(lobId);
423 if (in == null) {
424 throw DbException.get(ErrorCode.OBJECT_CLOSED);
425 }
426 }
427 long offset = transfer.readLong();
428 if (in.getPos() != offset) {
429 LobStorage lobStorage = session.getDataHandler().getLobStorage();
430 InputStream lobIn = lobStorage.getInputStream(lobId, hmac, -1);
431 in = new CachedInputStream(lobIn);
432 lobs.put(lobId, in);
433 lobIn.skip(offset);
434 }
435 int length = transfer.readInt();
436 // limit the buffer size
437 length = Math.min(16 * Constants.IO_BUFFER_SIZE, length);
438 transfer.writeInt(SessionRemote.STATUS_OK);
439 byte[] buff = new byte[length];
440 length = IOUtils.readFully(in, buff, 0, length);
441 transfer.writeInt(length);
442 transfer.writeBytes(buff, 0, length);
443 transfer.flush();
444 break;
445 }
446 default:
447 trace("Unknown operation: " + operation);
448 closeSession();
449 close();
450 }
451 }
452
453 private int getState(int oldModificationId) {
454 if (session.getModificationId() == oldModificationId) {
455 return SessionRemote.STATUS_OK;
456 }
457 return SessionRemote.STATUS_OK_STATE_CHANGED;
458 }
459
460 private void sendRow(ResultInterface result) throws IOException {
461 if (result.next()) {
462 transfer.writeBoolean(true);
463 Value[] v = result.currentRow();
464 for (int i = 0; i < result.getVisibleColumnCount(); i++) {
465 if (clientVersion >= Constants.TCP_PROTOCOL_VERSION_12) {
466 transfer.writeValue(v[i]);
467 } else {
468 writeValue(v[i]);
469 }
470 }
471 } else {
472 transfer.writeBoolean(false);
473 }
474 }
475
476 private void writeValue(Value v) throws IOException {
477 if (v.getType() == Value.CLOB || v.getType() == Value.BLOB) {
478 if (v instanceof ValueLobDb) {
479 ValueLobDb lob = (ValueLobDb) v;
480 if (lob.isStored()) {
481 long id = lob.getLobId();
482 lobs.put(id, new CachedInputStream(null));
483 }
484 }
485 }
486 transfer.writeValue(v);
487 }
488
489 void setThread(Thread thread) {
490 this.thread = thread;
491 }
492
493 Thread getThread() {
494 return thread;
495 }
496
497 /**
498 * Cancel a running statement.
499 *
500 * @param targetSessionId the session id
501 * @param statementId the statement to cancel
502 */
503 void cancelStatement(String targetSessionId, int statementId) {
504 if (StringUtils.equals(targetSessionId, this.sessionId)) {
505 Command cmd = (Command) cache.getObject(statementId, false);
506 cmd.cancel();
507 }
508 }
509
510 /**
511 * An input stream with a position.
512 */
513 static class CachedInputStream extends FilterInputStream {
514
515 private static final ByteArrayInputStream DUMMY = new ByteArrayInputStream(new byte[0]);
516 private long pos;
517
518 CachedInputStream(InputStream in) {
519 super(in == null ? DUMMY : in);
520 if (in == null) {
521 pos = -1;
522 }
523 }
524
525 public int read(byte[] buff, int off, int len) throws IOException {
526 len = super.read(buff, off, len);
527 if (len > 0) {
528 pos += len;
529 }
530 return len;
531 }
532
533 public int read() throws IOException {
534 int x = in.read();
535 if (x >= 0) {
536 pos++;
537 }
538 return x;
539 }
540
541 public long skip(long n) throws IOException {
542 n = super.skip(n);
543 if (n > 0) {
544 pos += n;
545 }
546 return n;
547 }
548
549 public long getPos() {
550 return pos;
551 }
552
553 }
554
555 }
556