File D:\maven-qcr-plugin-testsunitaires\qcr4java\h2-1.3.170-sources\target\site\qc\lecode\engine\SessionRemote.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.engine;
8
9 import java.io.IOException;
10 import java.net.Socket;
11 import java.sql.Connection;
12 import java.util.ArrayList;
13 import org.h2.api.DatabaseEventListener;
14 import org.h2.command.CommandInterface;
15 import org.h2.command.CommandRemote;
16 import org.h2.command.dml.SetTypes;
17 import org.h2.constant.ErrorCode;
18 import org.h2.constant.SysProperties;
19 import org.h2.jdbc.JdbcSQLException;
20 import org.h2.message.DbException;
21 import org.h2.message.Trace;
22 import org.h2.message.TraceSystem;
23 import org.h2.store.DataHandler;
24 import org.h2.store.FileStore;
25 import org.h2.store.LobStorage;
26 import org.h2.store.fs.FileUtils;
27 import org.h2.util.MathUtils;
28 import org.h2.util.NetUtils;
29 import org.h2.util.New;
30 import org.h2.util.SmallLRUCache;
31 import org.h2.util.StringUtils;
32 import org.h2.util.TempFileDeleter;
33 import org.h2.util.Utils;
34 import org.h2.value.Transfer;
35
36 /**
37 * The client side part of a session when using the server mode. This object
38 * communicates with a Session on the server side.
39 */
40 public class SessionRemote extends SessionWithState implements DataHandler {
41
42 public static final int SESSION_PREPARE = 0;
43 public static final int SESSION_CLOSE = 1;
44 public static final int COMMAND_EXECUTE_QUERY = 2;
45 public static final int COMMAND_EXECUTE_UPDATE = 3;
46 public static final int COMMAND_CLOSE = 4;
47 public static final int RESULT_FETCH_ROWS = 5;
48 public static final int RESULT_RESET = 6;
49 public static final int RESULT_CLOSE = 7;
50 public static final int COMMAND_COMMIT = 8;
51 public static final int CHANGE_ID = 9;
52 public static final int COMMAND_GET_META_DATA = 10;
53 public static final int SESSION_PREPARE_READ_PARAMS = 11;
54 public static final int SESSION_SET_ID = 12;
55 public static final int SESSION_CANCEL_STATEMENT = 13;
56 public static final int SESSION_CHECK_KEY = 14;
57 public static final int SESSION_SET_AUTOCOMMIT = 15;
58 public static final int SESSION_UNDO_LOG_POS = 16;
59 public static final int LOB_READ = 17;
60
61 public static final int STATUS_ERROR = 0;
62 public static final int STATUS_OK = 1;
63 public static final int STATUS_CLOSED = 2;
64 public static final int STATUS_OK_STATE_CHANGED = 3;
65
66 private static SessionFactory sessionFactory;
67
68 private TraceSystem traceSystem;
69 private Trace trace;
70 private ArrayList<Transfer> transferList = New.arrayList();
71 private int nextId;
72 private boolean autoCommit = true;
73 private CommandInterface autoCommitFalse, autoCommitTrue;
74 private ConnectionInfo connectionInfo;
75 private String databaseName;
76 private String cipher;
77 private byte[] fileEncryptionKey;
78 private final Object lobSyncObject = new Object();
79 private String sessionId;
80 private int clientVersion;
81 private boolean autoReconnect;
82 private int lastReconnect;
83 private SessionInterface embedded;
84 private DatabaseEventListener eventListener;
85 private LobStorage lobStorage;
86 private boolean cluster;
87
88 public SessionRemote(ConnectionInfo ci) {
89 this.connectionInfo = ci;
90 }
91
92 private Transfer initTransfer(ConnectionInfo ci, String db, String server) throws IOException {
93 Socket socket = NetUtils.createSocket(server, Constants.DEFAULT_TCP_PORT, ci.isSSL());
94 Transfer trans = new Transfer(this);
95 trans.setSocket(socket);
96 trans.setSSL(ci.isSSL());
97 trans.init();
98 trans.writeInt(Constants.TCP_PROTOCOL_VERSION_6);
99 trans.writeInt(Constants.TCP_PROTOCOL_VERSION_12);
100 trans.writeString(db);
101 trans.writeString(ci.getOriginalURL());
102 trans.writeString(ci.getUserName());
103 trans.writeBytes(ci.getUserPasswordHash());
104 trans.writeBytes(ci.getFilePasswordHash());
105 String[] keys = ci.getKeys();
106 trans.writeInt(keys.length);
107 for (String key : keys) {
108 trans.writeString(key).writeString(ci.getProperty(key));
109 }
110 try {
111 done(trans);
112 clientVersion = trans.readInt();
113 trans.setVersion(clientVersion);
114 trans.writeInt(SessionRemote.SESSION_SET_ID);
115 trans.writeString(sessionId);
116 done(trans);
117 } catch (DbException e) {
118 trans.close();
119 throw e;
120 }
121 autoCommit = true;
122 return trans;
123 }
124
125 public int getUndoLogPos() {
126 if (clientVersion < Constants.TCP_PROTOCOL_VERSION_10) {
127 return 1;
128 }
129 for (int i = 0, count = 0; i < transferList.size(); i++) {
130 Transfer transfer = transferList.get(i);
131 try {
132 traceOperation("SESSION_UNDO_LOG_POS", 0);
133 transfer.writeInt(SessionRemote.SESSION_UNDO_LOG_POS);
134 done(transfer);
135 return transfer.readInt();
136 } catch (IOException e) {
137 removeServer(e, i--, ++count);
138 }
139 }
140 return 1;
141 }
142
143 public void cancel() {
144 // this method is called when closing the connection
145 // the statement that is currently running is not canceled in this case
146 // however Statement.cancel is supported
147 }
148
149 /**
150 * Cancel the statement with the given id.
151 *
152 * @param id the statement id
153 */
154 public void cancelStatement(int id) {
155 for (Transfer transfer : transferList) {
156 try {
157 Transfer trans = transfer.openNewConnection();
158 trans.init();
159 trans.writeInt(clientVersion);
160 trans.writeInt(clientVersion);
161 trans.writeString(null);
162 trans.writeString(null);
163 trans.writeString(sessionId);
164 trans.writeInt(SessionRemote.SESSION_CANCEL_STATEMENT);
165 trans.writeInt(id);
166 trans.close();
167 } catch (IOException e) {
168 trace.debug(e, "could not cancel statement");
169 }
170 }
171 }
172
173 private void checkClusterDisableAutoCommit(String serverList) {
174 if (autoCommit && transferList.size() > 1) {
175 setAutoCommitSend(false);
176 CommandInterface c = prepareCommand("SET CLUSTER " + serverList, Integer.MAX_VALUE);
177 // this will set autoCommit to false
178 c.executeUpdate();
179 // so we need to switch it on
180 autoCommit = true;
181 cluster = true;
182 }
183 }
184
185 public boolean getAutoCommit() {
186 return autoCommit;
187 }
188
189 public void setAutoCommit(boolean autoCommit) {
190 if (!cluster) {
191 setAutoCommitSend(autoCommit);
192 }
193 this.autoCommit = autoCommit;
194 }
195
196 public void setAutoCommitFromServer(boolean autoCommit) {
197 if (cluster) {
198 if (autoCommit) {
199 // the user executed SET AUTOCOMMIT TRUE
200 setAutoCommitSend(false);
201 this.autoCommit = true;
202 }
203 } else {
204 this.autoCommit = autoCommit;
205 }
206 }
207
208 private void setAutoCommitSend(boolean autoCommit) {
209 if (clientVersion >= Constants.TCP_PROTOCOL_VERSION_8) {
210 for (int i = 0, count = 0; i < transferList.size(); i++) {
211 Transfer transfer = transferList.get(i);
212 try {
213 traceOperation("SESSION_SET_AUTOCOMMIT", autoCommit ? 1 : 0);
214 transfer.writeInt(SessionRemote.SESSION_SET_AUTOCOMMIT).writeBoolean(autoCommit);
215 done(transfer);
216 } catch (IOException e) {
217 removeServer(e, i--, ++count);
218 }
219 }
220 } else {
221 if (autoCommit) {
222 if (autoCommitTrue == null) {
223 autoCommitTrue = prepareCommand("SET AUTOCOMMIT TRUE", Integer.MAX_VALUE);
224 }
225 autoCommitTrue.executeUpdate();
226 } else {
227 if (autoCommitFalse == null) {
228 autoCommitFalse = prepareCommand("SET AUTOCOMMIT FALSE", Integer.MAX_VALUE);
229 }
230 autoCommitFalse.executeUpdate();
231 }
232 }
233 }
234
235 /**
236 * Calls COMMIT if the session is in cluster mode.
237 */
238 public void autoCommitIfCluster() {
239 if (autoCommit && cluster) {
240 // server side auto commit is off because of race conditions
241 // (update set id=1 where id=0, but update set id=2 where id=0 is
242 // faster)
243 for (int i = 0, count = 0; i < transferList.size(); i++) {
244 Transfer transfer = transferList.get(i);
245 try {
246 traceOperation("COMMAND_COMMIT", 0);
247 transfer.writeInt(SessionRemote.COMMAND_COMMIT);
248 done(transfer);
249 } catch (IOException e) {
250 removeServer(e, i--, ++count);
251 }
252 }
253 }
254 }
255
256 private String getFilePrefix(String dir) {
257 StringBuilder buff = new StringBuilder(dir);
258 buff.append('/');
259 for (int i = 0; i < databaseName.length(); i++) {
260 char ch = databaseName.charAt(i);
261 if (Character.isLetterOrDigit(ch)) {
262 buff.append(ch);
263 } else {
264 buff.append('_');
265 }
266 }
267 return buff.toString();
268 }
269
270 public int getPowerOffCount() {
271 return 0;
272 }
273
274 public void setPowerOffCount(int count) {
275 throw DbException.getUnsupportedException("remote");
276 }
277
278 /**
279 * Open a new (remote or embedded) session.
280 *
281 * @param openNew whether to open a new session in any case
282 * @return the session
283 */
284 public SessionInterface connectEmbeddedOrServer(boolean openNew) {
285 ConnectionInfo ci = connectionInfo;
286 if (ci.isRemote()) {
287 connectServer(ci);
288 return this;
289 }
290 // create the session using reflection,
291 // so that the JDBC layer can be compiled without it
292 boolean autoServerMode = Boolean.parseBoolean(ci.getProperty("AUTO_SERVER", "false"));
293 ConnectionInfo backup = null;
294 try {
295 if (autoServerMode) {
296 backup = (ConnectionInfo) ci.clone();
297 connectionInfo = (ConnectionInfo) ci.clone();
298 }
299 if (openNew) {
300 ci.setProperty("OPEN_NEW", "true");
301 }
302 if (sessionFactory == null) {
303 sessionFactory = (SessionFactory) Class.forName("org.h2.engine.Engine").newInstance();
304 }
305 return sessionFactory.createSession(ci);
306 } catch (Exception re) {
307 DbException e = DbException.convert(re);
308 if (e.getErrorCode() == ErrorCode.DATABASE_ALREADY_OPEN_1) {
309 if (autoServerMode) {
310 String serverKey = ((JdbcSQLException) e.getSQLException()).getSQL();
311 if (serverKey != null) {
312 backup.setServerKey(serverKey);
313 // OPEN_NEW must be removed now, otherwise
314 // opening a session with AUTO_SERVER fails
315 // if another connection is already open
316 backup.removeProperty("OPEN_NEW", null);
317 connectServer(backup);
318 return this;
319 }
320 }
321 }
322 throw e;
323 }
324 }
325
326 private void connectServer(ConnectionInfo ci) {
327 String name = ci.getName();
328 if (name.startsWith("//")) {
329 name = name.substring("//".length());
330 }
331 int idx = name.indexOf('/');
332 if (idx < 0) {
333 throw ci.getFormatException();
334 }
335 databaseName = name.substring(idx + 1);
336 String server = name.substring(0, idx);
337 traceSystem = new TraceSystem(null);
338 String traceLevelFile = ci.getProperty(SetTypes.TRACE_LEVEL_FILE, null);
339 if (traceLevelFile != null) {
340 int level = Integer.parseInt(traceLevelFile);
341 String prefix = getFilePrefix(SysProperties.CLIENT_TRACE_DIRECTORY);
342 try {
343 traceSystem.setLevelFile(level);
344 if (level > 0) {
345 String file = FileUtils.createTempFile(prefix, Constants.SUFFIX_TRACE_FILE, false, false);
346 traceSystem.setFileName(file);
347 }
348 } catch (IOException e) {
349 throw DbException.convertIOException(e, prefix);
350 }
351 }
352 String traceLevelSystemOut = ci.getProperty(SetTypes.TRACE_LEVEL_SYSTEM_OUT, null);
353 if (traceLevelSystemOut != null) {
354 int level = Integer.parseInt(traceLevelSystemOut);
355 traceSystem.setLevelSystemOut(level);
356 }
357 trace = traceSystem.getTrace(Trace.JDBC);
358 String serverList = null;
359 if (server.indexOf(',') >= 0) {
360 serverList = StringUtils.quoteStringSQL(server);
361 ci.setProperty("CLUSTER", Constants.CLUSTERING_ENABLED);
362 }
363 autoReconnect = Boolean.parseBoolean(ci.getProperty("AUTO_RECONNECT", "false"));
364 // AUTO_SERVER implies AUTO_RECONNECT
365 boolean autoServer = Boolean.parseBoolean(ci.getProperty("AUTO_SERVER", "false"));
366 if (autoServer && serverList != null) {
367 throw DbException.getUnsupportedException("autoServer && serverList != null");
368 }
369 autoReconnect |= autoServer;
370 if (autoReconnect) {
371 String className = ci.getProperty("DATABASE_EVENT_LISTENER");
372 if (className != null) {
373 className = StringUtils.trim(className, true, true, "'");
374 try {
375 eventListener = (DatabaseEventListener) Utils.loadUserClass(className).newInstance();
376 } catch (Throwable e) {
377 throw DbException.convert(e);
378 }
379 }
380 }
381 cipher = ci.getProperty("CIPHER");
382 if (cipher != null) {
383 fileEncryptionKey = MathUtils.secureRandomBytes(32);
384 }
385 String[] servers = StringUtils.arraySplit(server, ',', true);
386 int len = servers.length;
387 transferList.clear();
388 sessionId = StringUtils.convertBytesToHex(MathUtils.secureRandomBytes(32));
389 // TODO cluster: support more than 2 connections
390 boolean switchOffCluster = false;
391 try {
392 for (int i = 0; i < len; i++) {
393 String s = servers[i];
394 try {
395 Transfer trans = initTransfer(ci, databaseName, s);
396 transferList.add(trans);
397 } catch (IOException e) {
398 if (len == 1) {
399 throw DbException.get(ErrorCode.CONNECTION_BROKEN_1, e, e + ": " + s);
400 }
401 switchOffCluster = true;
402 }
403 }
404 checkClosed();
405 if (switchOffCluster) {
406 switchOffCluster();
407 }
408 checkClusterDisableAutoCommit(serverList);
409 } catch (DbException e) {
410 traceSystem.close();
411 throw e;
412 }
413 }
414
415 private void switchOffCluster() {
416 CommandInterface ci = prepareCommand("SET CLUSTER ''", Integer.MAX_VALUE);
417 ci.executeUpdate();
418 }
419
420 /**
421 * Remove a server from the list of cluster nodes and disables the cluster
422 * mode.
423 *
424 * @param e the exception (used for debugging)
425 * @param i the index of the server to remove
426 * @param count the retry count index
427 */
428 public void removeServer(IOException e, int i, int count) {
429 trace.debug(e, "removing server because of exception");
430 transferList.remove(i);
431 if (transferList.size() == 0 && autoReconnect(count)) {
432 return;
433 }
434 checkClosed();
435 switchOffCluster();
436 }
437
438 public synchronized CommandInterface prepareCommand(String sql, int fetchSize) {
439 checkClosed();
440 return new CommandRemote(this, transferList, sql, fetchSize);
441 }
442
443 /**
444 * Automatically re-connect if necessary and if configured to do so.
445 *
446 * @param count the retry count index
447 * @return true if reconnected
448 */
449 private boolean autoReconnect(int count) {
450 if (!isClosed()) {
451 return false;
452 }
453 if (!autoReconnect) {
454 return false;
455 }
456 if (!cluster && !autoCommit) {
457 return false;
458 }
459 if (count > SysProperties.MAX_RECONNECT) {
460 return false;
461 }
462 lastReconnect++;
463 while (true) {
464 try {
465 embedded = connectEmbeddedOrServer(false);
466 break;
467 } catch (DbException e) {
468 if (e.getErrorCode() != ErrorCode.DATABASE_IS_IN_EXCLUSIVE_MODE) {
469 throw e;
470 }
471 // exclusive mode: re-try endlessly
472 try {
473 Thread.sleep(500);
474 } catch (Exception e2) {
475 // ignore
476 }
477 }
478 }
479 if (embedded == this) {
480 // connected to a server somewhere else
481 embedded = null;
482 } else {
483 // opened an embedded connection now -
484 // must connect to this database in server mode
485 // unfortunately
486 connectEmbeddedOrServer(true);
487 }
488 recreateSessionState();
489 if (eventListener != null) {
490 eventListener.setProgress(DatabaseEventListener.STATE_RECONNECTED, databaseName, count,
491 SysProperties.MAX_RECONNECT);
492 }
493 return true;
494 }
495
496 /**
497 * Check if this session is closed and throws an exception if so.
498 *
499 * @throws DbException if the session is closed
500 */
501 public void checkClosed() {
502 if (isClosed()) {
503 throw DbException.get(ErrorCode.CONNECTION_BROKEN_1, "session closed");
504 }
505 }
506
507 public void close() {
508 RuntimeException closeError = null;
509 if (transferList != null) {
510 synchronized (this) {
511 for (Transfer transfer : transferList) {
512 try {
513 traceOperation("SESSION_CLOSE", 0);
514 transfer.writeInt(SessionRemote.SESSION_CLOSE);
515 done(transfer);
516 transfer.close();
517 } catch (RuntimeException e) {
518 trace.error(e, "close");
519 closeError = e;
520 } catch (Exception e) {
521 trace.error(e, "close");
522 }
523 }
524 }
525 transferList = null;
526 }
527 traceSystem.close();
528 if (embedded != null) {
529 embedded.close();
530 embedded = null;
531 }
532 if (closeError != null) {
533 throw closeError;
534 }
535 }
536
537 public Trace getTrace() {
538 return traceSystem.getTrace(Trace.JDBC);
539 }
540
541 public int getNextId() {
542 return nextId++;
543 }
544
545 public int getCurrentId() {
546 return nextId;
547 }
548
549 /**
550 * Called to flush the output after data has been sent to the server and
551 * just before receiving data. This method also reads the status code from
552 * the server and throws any exception the server sent.
553 *
554 * @param transfer the transfer object
555 * @throws DbException if the server sent an exception
556 * @throws IOException if there is a communication problem between client
557 * and server
558 */
559 public void done(Transfer transfer) throws IOException {
560 transfer.flush();
561 int status = transfer.readInt();
562 if (status == STATUS_ERROR) {
563 String sqlstate = transfer.readString();
564 String message = transfer.readString();
565 String sql = transfer.readString();
566 int errorCode = transfer.readInt();
567 String stackTrace = transfer.readString();
568 JdbcSQLException s = new JdbcSQLException(message, sql, sqlstate, errorCode, null, stackTrace);
569 if (errorCode == ErrorCode.CONNECTION_BROKEN_1) {
570 // allow re-connect
571 IOException e = new IOException(s.toString());
572 e.initCause(s);
573 throw e;
574 }
575 throw DbException.convert(s);
576 } else if (status == STATUS_CLOSED) {
577 transferList = null;
578 } else if (status == STATUS_OK_STATE_CHANGED) {
579 sessionStateChanged = true;
580 } else if (status == STATUS_OK) {
581 // ok
582 } else {
583 throw DbException.get(ErrorCode.CONNECTION_BROKEN_1, "unexpected status " + status);
584 }
585 }
586
587 /**
588 * Returns true if the connection was opened in cluster mode.
589 *
590 * @return true if it is
591 */
592 public boolean isClustered() {
593 return cluster;
594 }
595
596 public boolean isClosed() {
597 return transferList == null || transferList.size() == 0;
598 }
599
600 /**
601 * Write the operation to the trace system if debug trace is enabled.
602 *
603 * @param operation the operation performed
604 * @param id the id of the operation
605 */
606 public void traceOperation(String operation, int id) {
607 if (trace.isDebugEnabled()) {
608 trace.debug("{0} {1}", operation, id);
609 }
610 }
611
612 public void checkPowerOff() {
613 // ok
614 }
615
616 public void checkWritingAllowed() {
617 // ok
618 }
619
620 public String getDatabasePath() {
621 return "";
622 }
623
624 public String getLobCompressionAlgorithm(int type) {
625 return null;
626 }
627
628 public int getMaxLengthInplaceLob() {
629 return SysProperties.LOB_CLIENT_MAX_SIZE_MEMORY;
630 }
631
632 public FileStore openFile(String name, String mode, boolean mustExist) {
633 if (mustExist && !FileUtils.exists(name)) {
634 throw DbException.get(ErrorCode.FILE_NOT_FOUND_1, name);
635 }
636 FileStore store;
637 if (cipher == null) {
638 store = FileStore.open(this, name, mode);
639 } else {
640 store = FileStore.open(this, name, mode, cipher, fileEncryptionKey, 0);
641 }
642 store.setCheckedWriting(false);
643 try {
644 store.init();
645 } catch (DbException e) {
646 store.closeSilently();
647 throw e;
648 }
649 return store;
650 }
651
652 public DataHandler getDataHandler() {
653 return this;
654 }
655
656 public Object getLobSyncObject() {
657 return lobSyncObject;
658 }
659
660 public SmallLRUCache<String, String[]> getLobFileListCache() {
661 return null;
662 }
663
664 public int getLastReconnect() {
665 return lastReconnect;
666 }
667
668 public TempFileDeleter getTempFileDeleter() {
669 return TempFileDeleter.getInstance();
670 }
671
672 public boolean isReconnectNeeded(boolean write) {
673 return false;
674 }
675
676 public SessionInterface reconnect(boolean write) {
677 return this;
678 }
679
680 public void afterWriting() {
681 // nothing to do
682 }
683
684 public LobStorage getLobStorage() {
685 if (lobStorage == null) {
686 lobStorage = new LobStorage(this);
687 }
688 return lobStorage;
689 }
690
691 public Connection getLobConnection() {
692 return null;
693 }
694
695 public synchronized int readLob(long lobId, byte[] hmac, long offset, byte[] buff, int off, int length) {
696 for (int i = 0, count = 0; i < transferList.size(); i++) {
697 Transfer transfer = transferList.get(i);
698 try {
699 traceOperation("LOB_READ", (int) lobId);
700 transfer.writeInt(SessionRemote.LOB_READ);
701 transfer.writeLong(lobId);
702 if (clientVersion >= Constants.TCP_PROTOCOL_VERSION_12) {
703 transfer.writeBytes(hmac);
704 }
705 transfer.writeLong(offset);
706 transfer.writeInt(length);
707 done(transfer);
708 length = transfer.readInt();
709 if (length <= 0) {
710 return length;
711 }
712 transfer.readBytes(buff, off, length);
713 return length;
714 } catch (IOException e) {
715 removeServer(e, i--, ++count);
716 }
717 }
718 return 1;
719 }
720
721 }
722