VirtualDatabaseWorkerThread
handles a connection with a C-JDBC driver.
VirtualDatabaseWorkerThread.java の 58 行で定義されています。
Public メソッド | |
VirtualDatabaseWorkerThread (Controller controller, VirtualDatabase vdb) | |
void | shutdown () |
void | run () |
String[] | retrieveClientData () |
long | getBytesRead () |
long | getBytesWritten () |
long | getTimeActive () |
long | getReadingSpeed () |
long | getWritingSpeed () |
String | getUser () |
Private メソッド | |
boolean | setRequestParameters (AbstractRequest request, String login, long tid, boolean transactionStarted) throws SQLException |
Private 変数 | |
boolean | isKilled = false |
VirtualDatabase | vdb |
final CommandCompleted | completed = new CommandCompleted() |
Trace | logger = null |
CJDBCInputStream | in = null |
CJDBCOutputStream | out = null |
VirtualDatabaseUser | user |
Controller | controller |
Static Private 変数 | |
final int | STREAM_GC_LIMIT = 5 |
int | threadID = 0 |
|
Creates a new
VirtualDatabaseWorkerThread.java の 104 行で定義されています。 参照先 org.objectweb.cjdbc.common.log.Trace.getLogger(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getName(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.logger, org.objectweb.cjdbc.controller.core.Controller.logger, と org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.threadID.
00105 { 00106 super("VirtualDatabaseWorkerThread"); 00107 this.vdb = vdb; 00108 this.controller = controller; 00109 try 00110 { 00111 this.logger = Trace 00112 .getLogger("org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread." 00113 + vdb.getName() + "." + (threadID++)); 00114 } 00115 catch (Exception e) 00116 { 00117 this.logger = vdb.logger; 00118 } 00119 } |
|
get bytes read
VirtualDatabaseWorkerThread.java の 833 行で定義されています。 参照先 org.objectweb.cjdbc.common.stream.CJDBCInputStream.getBytesRead(), と org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.in.
00834 { 00835 return in.getBytesRead(); 00836 } |
|
get bytes written
VirtualDatabaseWorkerThread.java の 843 行で定義されています。 参照先 org.objectweb.cjdbc.common.stream.CJDBCOutputStream.getBytesWritten(), と org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.out.
00844 { 00845 return out.getBytesWritten(); 00846 } |
|
get reading speed, WARNING! This is approximate
VirtualDatabaseWorkerThread.java の 863 行で定義されています。 参照先 org.objectweb.cjdbc.common.stream.CJDBCInputStream.getSpeed(), と org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.in.
00864 { 00865 return in.getSpeed(); 00866 } |
|
get time active
VirtualDatabaseWorkerThread.java の 853 行で定義されています。 参照先 org.objectweb.cjdbc.common.stream.CJDBCInputStream.getDateCreated(), と org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.in.
00854 { 00855 return ((System.currentTimeMillis() - in.getDateCreated()) / 1000); 00856 } |
|
VirtualDatabaseWorkerThread.java の 881 行で定義されています。 参照先 org.objectweb.cjdbc.controller.authentication.AbstractDatabaseUser.getLogin(). 参照元 org.objectweb.cjdbc.common.monitor.client.AbstractClientDataCollector.checkClientIndex(), と org.objectweb.cjdbc.common.monitor.client.AbstractClientDataCollector.setClientIndex().
00882 { 00883 return user.getLogin(); 00884 } |
|
get writing speed, WARNING! This is approximate
VirtualDatabaseWorkerThread.java の 873 行で定義されています。 参照先 org.objectweb.cjdbc.common.stream.CJDBCOutputStream.getSpeed(), と org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.out.
00874 { 00875 return out.getSpeed(); 00876 } |
|
|
Gets a connection from the connection queue and process it. VirtualDatabaseWorkerThread.java の 159 行で定義されています。 参照先 org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.addIdleThread(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.begin(), org.objectweb.cjdbc.common.stream.CJDBCInputStream.close(), org.objectweb.cjdbc.common.stream.CJDBCOutputStream.close(), org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet.closeResultSet(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.commit(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.completed, org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.currentNbOfThreads, org.objectweb.cjdbc.common.log.Trace.debug(), org.objectweb.cjdbc.common.log.Trace.error(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.execReadRequest(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.execReadStoredProcedure(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.execWriteRequest(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.execWriteRequestWithKeys(), org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet.fetchData(), org.objectweb.cjdbc.common.stream.CJDBCOutputStream.flush(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getActiveThreads(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getAuthenticationManager(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getBlobFilter(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getCatalogs(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getColumns(), org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet.getCursorName(), org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet.getData(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getDatabaseName(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getDatabaseProductName(), org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet.getFields(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getMaxThreadIdleTime(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getMetaData(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getName(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getPendingConnections(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getPrimaryKeys(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getProcedureColumns(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getProcedures(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.getRequestManager(), org.objectweb.cjdbc.controller.requestmanager.RequestManager.getRequiredParsingGranularity(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getSchemas(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getTablePrivileges(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getTables(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseMetaData.getTableTypes(), org.objectweb.cjdbc.controller.core.Controller.getVirtualDatabase(), org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet.hasMoreData, org.objectweb.cjdbc.controller.core.Controller.hasVirtualDatabase(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.in, org.objectweb.cjdbc.common.log.Trace.isDebugEnabled(), org.objectweb.cjdbc.controller.authentication.AuthenticationManager.isValidVirtualUser(), org.objectweb.cjdbc.controller.core.Controller.listVirtualDatabases(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.minNbOfThreads, org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.out, org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.poolConnectionThreads, org.objectweb.cjdbc.common.stream.CJDBCInputStream.readInt(), org.objectweb.cjdbc.common.stream.CJDBCInputStream.readObject(), org.objectweb.cjdbc.common.stream.CJDBCInputStream.readUTF(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.removeCurrentNbOfThread(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.removeIdleThread(), org.objectweb.cjdbc.common.stream.CJDBCOutputStream.reset(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.rollback(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.setRequestParameters(), org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.STREAM_GC_LIMIT, org.objectweb.cjdbc.common.log.Trace.warn(), org.objectweb.cjdbc.common.stream.CJDBCOutputStream.writeBoolean(), org.objectweb.cjdbc.common.stream.CJDBCOutputStream.writeObject(), と org.objectweb.cjdbc.common.stream.CJDBCOutputStream.writeUTF().
00160 { 00161 ArrayList vdbActiveThreads = vdb.getActiveThreads(); 00162 ArrayList vdbPendingQueue = vdb.getPendingConnections(); 00163 // List of open ResultSets for streaming. This is not synchronized since the 00164 // connection does only handle one request at a time 00165 HashMap streamedResultSet = new HashMap(); 00166 boolean isActive = true; 00167 00168 if (vdbActiveThreads == null) 00169 { 00170 logger 00171 .error("Got null active threads queue in VirtualDatabaseWorkerThread"); 00172 isKilled = true; 00173 } 00174 if (vdbPendingQueue == null) 00175 { 00176 logger.error("Got null connection queue in VirtualDatabaseWorkerThread"); 00177 isKilled = true; 00178 } 00179 00180 // Main loop 00181 while (!isKilled) 00182 { 00183 // Get a connection from the pending queue 00184 synchronized (vdbPendingQueue) 00185 { 00186 while (vdbPendingQueue.isEmpty()) 00187 { 00188 if (!vdb.poolConnectionThreads) 00189 { // User does not want thread pooling, kill this thread! 00190 isKilled = true; 00191 break; 00192 } 00193 boolean timeout = false; 00194 try 00195 { 00196 if (isActive) 00197 { 00198 isActive = false; 00199 // Remove ourselves from the active thread list 00200 synchronized (vdbActiveThreads) 00201 { 00202 vdbActiveThreads.remove(this); 00203 vdb.addIdleThread(); 00204 } 00205 } 00206 long before = System.currentTimeMillis(); 00207 vdbPendingQueue.wait(vdb.getMaxThreadIdleTime()); 00208 long now = System.currentTimeMillis(); 00209 // Check if timeout has expired 00210 timeout = now - before >= vdb.getMaxThreadIdleTime(); 00211 } 00212 catch (InterruptedException e) 00213 { 00214 logger.warn("VirtualDatabaseWorkerThread wait() interrupted"); 00215 } 00216 if (timeout && vdbPendingQueue.isEmpty()) 00217 { 00218 if (vdb.currentNbOfThreads > vdb.minNbOfThreads) 00219 { // We have enough threads, kill this one 00220 isKilled = true; 00221 break; 00222 } 00223 } 00224 } 00225 00226 if (isKilled) 00227 { // Cleaning up 00228 synchronized (vdbActiveThreads) 00229 { // Remove ourselves from the appropriate thread list 00230 if (isActive) 00231 { 00232 vdbActiveThreads.remove(this); 00233 vdb.removeCurrentNbOfThread(); 00234 } 00235 else 00236 vdb.removeIdleThread(); 00237 } 00238 // Get out of the while loop 00239 continue; 00240 } 00241 00242 // Get a connection 00243 try 00244 { 00245 in = (CJDBCInputStream) vdbPendingQueue.remove(0); 00246 out = (CJDBCOutputStream) vdbPendingQueue.remove(0); 00247 } 00248 catch (Exception e) 00249 { 00250 logger.error("Error while getting streams from connection"); 00251 continue; 00252 } 00253 00254 synchronized (vdbActiveThreads) 00255 { 00256 if (!isActive) 00257 { 00258 vdb.removeIdleThread(); 00259 isActive = true; 00260 // Add this thread to the active thread list 00261 vdbActiveThreads.add(this); 00262 } 00263 } 00264 } 00265 00266 // Handle connection 00267 // Read the user information and check authentication 00268 boolean success = false; 00269 String login; 00270 try 00271 { 00272 login = in.readUTF(); 00273 String password = in.readUTF(); 00274 user = new VirtualDatabaseUser(login, password); 00275 00276 if (vdb.getAuthenticationManager().isValidVirtualUser(user)) 00277 { // Authentication ok, send if SQL skeleton is needed 00278 boolean needSkeleton = vdb.getRequestManager() 00279 .getRequiredParsingGranularity() != ParsingGranularities.NO_PARSING; 00280 out.writeObject(new Boolean(needSkeleton)); 00281 out.writeObject(vdb.getBlobFilter()); 00282 out.flush(); 00283 success = true; 00284 00285 if (logger.isDebugEnabled()) 00286 logger.debug("Login accepted for " + login); 00287 } 00288 else 00289 { // Authentication failed, close the connection 00290 if (logger.isDebugEnabled()) 00291 logger.debug("Authentication failed for user '" + login + "'"); 00292 continue; 00293 } 00294 } 00295 catch (OptionalDataException e) 00296 { 00297 logger.error("Protocol error while expecting user authentication (" + e 00298 + ")"); 00299 continue; 00300 } 00301 catch (IOException e) 00302 { 00303 logger.error("I/O error during user authentication (" + e + ")"); 00304 continue; 00305 } 00306 finally 00307 { 00308 if (!success) 00309 { 00310 try 00311 { 00312 out.close(); 00313 in.close(); 00314 } 00315 catch (IOException ignore) 00316 { 00317 } 00318 } 00319 } 00320 00321 // Ok, let's process the queries on this connection 00322 long currentTid = 0; 00323 boolean transactionStarted = false; 00324 boolean closed = false; 00325 int objectsOnStream = 0; 00326 int command; 00327 while (!closed && !isKilled) 00328 { 00329 try 00330 { 00331 // Get the query 00332 command = in.readInt(); 00333 00334 // Process it 00335 switch (command) 00336 { 00337 case Commands.ExecReadRequest : 00338 if (logger.isDebugEnabled()) 00339 logger.debug("ExecReadRequest command"); 00340 SelectRequest select = (SelectRequest) in.readObject(); 00341 transactionStarted = setRequestParameters(select, login, 00342 currentTid, transactionStarted); 00343 if (!transactionStarted) 00344 currentTid = 0; 00345 ControllerResultSet crs = vdb.execReadRequest(select); 00346 if (crs.hasMoreData()) 00347 streamedResultSet.put(crs.getCursorName(), crs); 00348 out.writeObject(crs.getFields()); 00349 out.writeObject(crs.getData()); 00350 out.writeBoolean(crs.hasMoreData()); 00351 if (crs.hasMoreData()) 00352 out.writeUTF(crs.getCursorName()); 00353 out.flush(); 00354 objectsOnStream++; 00355 break; 00356 case Commands.ExecWriteRequest : 00357 if (logger.isDebugEnabled()) 00358 logger.debug("ExecWriteRequest command"); 00359 AbstractWriteRequest write = (AbstractWriteRequest) in 00360 .readObject(); 00361 transactionStarted = setRequestParameters(write, login, 00362 currentTid, transactionStarted); 00363 if (!transactionStarted) 00364 currentTid = 0; 00365 out.writeObject(new Integer(vdb.execWriteRequest(write))); 00366 out.flush(); 00367 objectsOnStream++; 00368 break; 00369 case Commands.ExecWriteRequestWithKeys : 00370 if (logger.isDebugEnabled()) 00371 logger.debug("ExecWriteRequestWithKeys command"); 00372 AbstractWriteRequest writeWithKeys = (AbstractWriteRequest) in 00373 .readObject(); 00374 transactionStarted = setRequestParameters(writeWithKeys, login, 00375 currentTid, transactionStarted); 00376 if (!transactionStarted) 00377 currentTid = 0; 00378 ControllerResultSet keys = vdb 00379 .execWriteRequestWithKeys(writeWithKeys); 00380 out.writeObject(keys.getFields()); 00381 out.writeObject(keys.getData()); 00382 out.writeBoolean(keys.hasMoreData()); 00383 if (keys.hasMoreData()) 00384 out.writeUTF(keys.getCursorName()); 00385 out.flush(); 00386 objectsOnStream++; 00387 break; 00388 case Commands.ExecReadStoredProcedure : 00389 if (logger.isDebugEnabled()) 00390 logger.debug("ExecReadStoredProcedure command"); 00391 StoredProcedure readProc = (StoredProcedure) in.readObject(); 00392 transactionStarted = setRequestParameters(readProc, login, 00393 currentTid, transactionStarted); 00394 if (!transactionStarted) 00395 currentTid = 0; 00396 ControllerResultSet sprs = vdb.execReadStoredProcedure(readProc); 00397 out.writeObject(sprs.getFields()); 00398 out.writeObject(sprs.getData()); 00399 out.writeBoolean(sprs.hasMoreData()); 00400 if (sprs.hasMoreData()) 00401 out.writeUTF(sprs.getCursorName()); 00402 out.flush(); 00403 objectsOnStream++; 00404 break; 00405 case Commands.ExecWriteStoredProcedure : 00406 if (logger.isDebugEnabled()) 00407 logger.debug("ExecWriteStoredProcedure command"); 00408 StoredProcedure writeProc = (StoredProcedure) in.readObject(); 00409 transactionStarted = setRequestParameters(writeProc, login, 00410 currentTid, transactionStarted); 00411 if (!transactionStarted) 00412 currentTid = 0; 00413 out.writeObject(new Integer(vdb 00414 .execWriteStoredProcedure(writeProc))); 00415 out.flush(); 00416 objectsOnStream++; 00417 break; 00418 case Commands.Begin : 00419 if (logger.isDebugEnabled()) 00420 logger.debug("Begin command"); 00421 currentTid = vdb.begin(login); 00422 out.writeObject(new Long(currentTid)); 00423 out.flush(); 00424 transactionStarted = true; 00425 break; 00426 case Commands.Commit : 00427 if (logger.isDebugEnabled()) 00428 logger.debug("Commit command"); 00429 vdb.commit(currentTid); 00430 currentTid = vdb.begin(login); 00431 out.writeObject(new Long(currentTid)); 00432 out.flush(); 00433 if (objectsOnStream > 0) 00434 { 00435 // Reset the outputstream else sent ResultSets cannot be GC 00436 out.reset(); 00437 objectsOnStream = 0; 00438 } 00439 break; 00440 case Commands.Rollback : 00441 if (logger.isDebugEnabled()) 00442 logger.debug("Rollback command"); 00443 vdb.rollback(currentTid); 00444 currentTid = vdb.begin(login); 00445 out.writeObject(new Long(currentTid)); 00446 out.flush(); 00447 if (objectsOnStream > 0) 00448 { 00449 // Reset the outputstream else sent ResultSets cannot be GC 00450 out.reset(); 00451 objectsOnStream = 0; 00452 } 00453 break; 00454 case Commands.GetVirtualDatabaseName : 00455 if (logger.isDebugEnabled()) 00456 logger.debug("GetVirtualDatabaseName command"); 00457 out.writeObject(vdb.getDatabaseName()); 00458 out.flush(); 00459 break; 00460 case Commands.GetDatabaseProductName : 00461 if (logger.isDebugEnabled()) 00462 logger.debug("GetDatabaseProductName command"); 00463 out.writeObject(vdb.getDatabaseProductName()); 00464 out.flush(); 00465 break; 00466 case Commands.GetControllerVersionNumber : 00467 if (logger.isDebugEnabled()) 00468 logger.debug("GetControllerVersionNumber command"); 00469 out.writeObject(Constants.VERSION); 00470 out.flush(); 00471 break; 00472 case Commands.DatabaseMetaDataGetTables : 00473 if (logger.isDebugEnabled()) 00474 logger.debug("DatabaseMetaDataGetTables command"); 00475 String tcatalog = (String) in.readObject(); 00476 String tschemaPattern = (String) in.readObject(); 00477 String ttableNamePattern = (String) in.readObject(); 00478 String[] ttypes = (String[]) in.readObject(); 00479 out.writeObject(vdb.getMetaData().getTables(tcatalog, 00480 tschemaPattern, ttableNamePattern, ttypes)); 00481 out.flush(); 00482 objectsOnStream++; 00483 break; 00484 case Commands.DatabaseMetaDataGetColumns : 00485 if (logger.isDebugEnabled()) 00486 logger.debug("DatabaseMetaDataGetColumns command"); 00487 String ccatalog = (String) in.readObject(); 00488 String cschemaPattern = (String) in.readObject(); 00489 String ctableNamePattern = (String) in.readObject(); 00490 String ccolumnNamePattern = (String) in.readObject(); 00491 out.writeObject(vdb.getMetaData().getColumns(ccatalog, 00492 cschemaPattern, ctableNamePattern, ccolumnNamePattern)); 00493 out.flush(); 00494 objectsOnStream++; 00495 break; 00496 case Commands.DatabaseMetaDataGetPrimaryKeys : 00497 if (logger.isDebugEnabled()) 00498 logger.debug("DatabaseMetaDataGetPrimaryKeys command"); 00499 String pcatalog = (String) in.readObject(); 00500 String pschemaPattern = (String) in.readObject(); 00501 String ptableNamePattern = (String) in.readObject(); 00502 out.writeObject(vdb.getMetaData().getPrimaryKeys(pcatalog, 00503 pschemaPattern, ptableNamePattern)); 00504 out.flush(); 00505 objectsOnStream++; 00506 break; 00507 case Commands.DatabaseMetaDataGetProcedures : 00508 if (logger.isDebugEnabled()) 00509 logger.debug("DatabaseMetaDataGetProcedures command"); 00510 String rcatalog = (String) in.readObject(); 00511 String rschemaPattern = (String) in.readObject(); 00512 String procedureNamePattern = (String) in.readObject(); 00513 out.writeObject(vdb.getMetaData().getProcedures(rcatalog, 00514 rschemaPattern, procedureNamePattern)); 00515 out.flush(); 00516 objectsOnStream++; 00517 break; 00518 case Commands.DatabaseMetaDataGetProcedureColumns : 00519 if (logger.isDebugEnabled()) 00520 logger.debug("DatabaseMetaDataGetProcedureColumns command"); 00521 String pccatalog = (String) in.readObject(); 00522 String pcschemaPattern = (String) in.readObject(); 00523 String pcprocedureNamePattern = (String) in.readObject(); 00524 String pccolumnNamePattern = (String) in.readObject(); 00525 out 00526 .writeObject(vdb.getMetaData().getProcedureColumns(pccatalog, 00527 pcschemaPattern, pcprocedureNamePattern, 00528 pccolumnNamePattern)); 00529 out.flush(); 00530 objectsOnStream++; 00531 break; 00532 case Commands.DatabaseMetaDataGetCatalogs : 00533 if (logger.isDebugEnabled()) 00534 logger.debug("DatabaseMetaDataGetCatalogs command"); 00535 ArrayList list = controller.listVirtualDatabases(); 00536 out.writeObject(vdb.getMetaData().getCatalogs(list)); 00537 out.flush(); 00538 objectsOnStream++; 00539 break; 00540 case Commands.DatabaseMetaDataGetCatalog : 00541 if (logger.isDebugEnabled()) 00542 logger.debug("DatabaseMetaDataGetCatalog command"); 00543 out.writeObject(vdb.getName()); 00544 out.flush(); 00545 objectsOnStream++; 00546 break; 00547 case Commands.DatabaseMetaDataGetTableTypes : 00548 if (logger.isDebugEnabled()) 00549 logger.debug("DatabaseMetaDataGetTable Types command"); 00550 out.writeObject(vdb.getMetaData().getTableTypes()); 00551 out.flush(); 00552 objectsOnStream++; 00553 break; 00554 case Commands.DatabaseMetaDataGetSchemas : 00555 if (logger.isDebugEnabled()) 00556 logger.debug("DatabaseMetaDataGetSchemas Types command"); 00557 out.writeObject(vdb.getMetaData().getSchemas()); 00558 out.flush(); 00559 objectsOnStream++; 00560 break; 00561 case Commands.DatabaseMetaDataGetTablePrivileges : 00562 if (logger.isDebugEnabled()) 00563 logger.debug("DatabaseMetaDataGetTablePrivileges command"); 00564 String tpcatalog = (String) in.readObject(); 00565 String tpschemaPattern = (String) in.readObject(); 00566 String tptablePattern = (String) in.readObject(); 00567 out.writeObject(vdb.getMetaData().getTablePrivileges(tpcatalog, 00568 tpschemaPattern, tptablePattern)); 00569 out.flush(); 00570 objectsOnStream++; 00571 break; 00572 case Commands.ConnectionSetCatalog : 00573 if (logger.isDebugEnabled()) 00574 logger.debug("ConnectionSetCatalog command"); 00575 String catalog = (String) in.readObject(); 00576 boolean change = controller.hasVirtualDatabase(catalog); 00577 if (change) 00578 { 00579 VirtualDatabase tempvdb = controller 00580 .getVirtualDatabase(catalog); 00581 if (!tempvdb.getAuthenticationManager() 00582 .isValidVirtualUser(user)) 00583 out 00584 .writeObject(new SQLException( 00585 "User authentication has failed for asked catalog. No change")); 00586 else 00587 { 00588 this.vdb = tempvdb; 00589 out.writeObject(new Boolean(true)); 00590 } 00591 } 00592 else 00593 out.writeObject(new Boolean(false)); 00594 out.flush(); 00595 objectsOnStream++; 00596 break; 00597 case Commands.Close : 00598 if (logger.isDebugEnabled()) 00599 logger.debug("Close command"); 00600 out.writeObject(completed); 00601 out.flush(); 00602 closed = true; 00603 break; 00604 case Commands.Reset : 00605 if (logger.isDebugEnabled()) 00606 logger.debug("Reset command"); 00607 00608 // Reset the outputstream else sent ResultSets cannot be GC 00609 out.reset(); 00610 00611 // Do the cleanup 00612 if (transactionStarted) 00613 { 00614 if (logger.isDebugEnabled()) 00615 logger.debug("Forcing transaction " + currentTid 00616 + " rollback"); 00617 try 00618 { 00619 vdb.rollback(currentTid); 00620 } 00621 catch (Exception e) 00622 { 00623 if (logger.isDebugEnabled()) 00624 logger.debug("Error during rollback of transaction " 00625 + currentTid + "(" + e + ")"); 00626 } 00627 } 00628 currentTid = 0; 00629 transactionStarted = false; 00630 break; 00631 case Commands.FetchNextResultSetRows : 00632 if (logger.isDebugEnabled()) 00633 logger.debug("FetchNextResultSetRows command"); 00634 00635 String cursorName = in.readUTF(); 00636 int fetchSize = in.readInt(); 00637 ControllerResultSet fetchCrs = (ControllerResultSet) streamedResultSet 00638 .get(cursorName); 00639 if (fetchCrs == null) 00640 { 00641 out.writeObject(new SQLException( 00642 "No valid ControllerResultSet to fetch data from")); 00643 out.flush(); 00644 } 00645 else 00646 { 00647 out.writeObject(fetchCrs.fetchData()); 00648 out.writeBoolean(fetchCrs.hasMoreData()); 00649 out.flush(); 00650 if (!fetchCrs.hasMoreData()) 00651 streamedResultSet.remove(cursorName); 00652 } 00653 break; 00654 case Commands.CloseRemoteResultSet : 00655 if (logger.isDebugEnabled()) 00656 logger.debug("CloseRemoteResultSet command"); 00657 00658 String cursor = in.readUTF(); 00659 ControllerResultSet crsToClose = (ControllerResultSet) streamedResultSet 00660 .remove(cursor); 00661 if (crsToClose == null) 00662 { 00663 out.writeObject(new SQLException( 00664 "No valid ControllerResultSet to close.")); 00665 out.flush(); 00666 } 00667 else 00668 { 00669 crsToClose.closeResultSet(); 00670 out.writeObject(completed); 00671 out.flush(); 00672 } 00673 break; 00674 default : 00675 logger.error("Unsupported protocol command: " + command); 00676 break; 00677 } 00678 00679 // Check if the stream needs to be reseted 00680 if (objectsOnStream > STREAM_GC_LIMIT) 00681 { // Reset the outputstream else sent ResultSets cannot be GC 00682 out.reset(); 00683 objectsOnStream = 0; 00684 } 00685 } 00686 catch (OptionalDataException e) 00687 { 00688 logger.warn("Protocol error (" + e + ")"); 00689 try 00690 { 00691 out.writeObject(e); 00692 out.flush(); 00693 } 00694 catch (IOException ignore) 00695 { 00696 } 00697 } 00698 catch (RemoteException e) 00699 { 00700 logger.warn("Error during command execution (" + e + ")"); 00701 try 00702 { 00703 out.writeObject(e); 00704 out.flush(); 00705 } 00706 catch (IOException ignore) 00707 { 00708 } 00709 } 00710 catch (EOFException e) 00711 { 00712 logger.warn("Client " + login + " closed connection with server"); 00713 closed = true; 00714 } 00715 catch (IOException e) 00716 { 00717 logger.warn("Closing connection with client " + login 00718 + " because of IOException.(" + e + ")"); 00719 closed = true; 00720 } 00721 catch (ClassNotFoundException e) 00722 { 00723 logger.error("Protocol error (" + e + ")"); 00724 try 00725 { 00726 out.writeObject(e); 00727 out.flush(); 00728 } 00729 catch (IOException ignore) 00730 { 00731 } 00732 } 00733 catch (SQLException e) 00734 { 00735 logger 00736 .warn("Error during command execution (" + e.getMessage() + ")"); 00737 try 00738 { 00739 out.writeObject(e); 00740 out.flush(); 00741 } 00742 catch (IOException ignore) 00743 { 00744 } 00745 } 00746 catch (RuntimeException e) 00747 { 00748 logger.warn("Runtime error during command execution (" 00749 + e.getMessage() + ")"); 00750 try 00751 { 00752 out.writeObject(new SQLException(e.getMessage())); 00753 out.flush(); 00754 } 00755 catch (IOException ignore) 00756 { 00757 } 00758 } 00759 } 00760 00761 // Do the cleanup 00762 if (transactionStarted) 00763 { 00764 if (logger.isDebugEnabled()) 00765 logger.debug("Forcing transaction " + currentTid + " rollback"); 00766 try 00767 { 00768 vdb.rollback(currentTid); 00769 } 00770 catch (Exception e) 00771 { 00772 if (logger.isDebugEnabled()) 00773 logger.debug("Error during rollback of transaction " + currentTid 00774 + "(" + e + ")"); 00775 } 00776 } 00777 if (!streamedResultSet.isEmpty()) 00778 { 00779 for (Iterator iter = streamedResultSet.values().iterator(); iter 00780 .hasNext();) 00781 { 00782 ControllerResultSet crs = (ControllerResultSet) iter.next(); 00783 crs.closeResultSet(); 00784 00785 } 00786 streamedResultSet.clear(); 00787 } 00788 try 00789 { 00790 in.close(); 00791 } 00792 catch (IOException ignore) 00793 { 00794 } 00795 try 00796 { 00797 out.close(); 00798 } 00799 catch (IOException ignore) 00800 { 00801 } 00802 } 00803 00804 if (logger.isDebugEnabled()) 00805 logger.debug("VirtualDatabaseWorkerThread terminating."); 00806 } |
|
Set the login and transaction id on the given request. If the request is autocommit and a transaction was started, the transaction is first commited to return in autocommit mode.
VirtualDatabaseWorkerThread.java の 142 行で定義されています。 参照先 org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase.commit(). 参照元 org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.run().
00144 { 00145 request.setLogin(login); 00146 if (request.isAutoCommit() && transactionStarted) 00147 { 00148 vdb.commit(tid); 00149 return false; 00150 } 00151 else 00152 request.setTransactionId(tid); 00153 return transactionStarted; 00154 } |
|
Shutdown this thread by setting VirtualDatabaseWorkerThread.java の 125 行で定義されています。
00126 { 00127 // Tell this thread to stop working gently. 00128 // This will cancel transaction if needed 00129 this.isKilled = true; 00130 } |
|
VirtualDatabaseWorkerThread.java の 82 行で定義されています。 参照元 org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.run(). |
|
VirtualDatabaseWorkerThread.java の 92 行で定義されています。 |
|
|
VirtualDatabaseWorkerThread.java の 77 行で定義されています。 |
|
Logger instance. VirtualDatabaseWorkerThread.java の 85 行で定義されています。 |
|
|
How many objects can be sent before resetting the output stream to allow those objects to be garbage collected. VirtualDatabaseWorkerThread.java の 72 行で定義されています。 参照元 org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabaseWorkerThread.run(). |
|
VirtualDatabaseWorkerThread.java の 74 行で定義されています。 |
|
VirtualDatabaseWorkerThread.java の 90 行で定義されています。 |
|
Virtual database instantiating this thread. VirtualDatabaseWorkerThread.java の 80 行で定義されています。 |