删除不必要的AsyncCallback子类，并在AsyncResult中处理异常情况

lealone-client/src/main/java/org/lealone/client/ClientBatchCommand.java

@@ -30,8 +30,8 @@
 import org.lealone.db.Session;
 import org.lealone.db.result.Result;
 import org.lealone.db.value.Value;
+import org.lealone.net.AsyncCallback;
 import org.lealone.net.Transfer;
-import org.lealone.net.VoidAsyncCallback;
 
 public class ClientBatchCommand implements Command {
     private ClientSession session;
@@ -132,7 +132,7 @@ public int executeUpdate() {
     }
 
     private void getResultAsync() throws IOException {
-        VoidAsyncCallback ac = new VoidAsyncCallback() {
+        AsyncCallback<Void> ac = new AsyncCallback<Void>() {
             @Override
             public void runInternal() {
                 try {

lealone-client/src/main/java/org/lealone/client/ClientCommand.java

@@ -26,9 +26,7 @@
 import org.lealone.db.result.Result;
 import org.lealone.db.value.Value;
 import org.lealone.net.AsyncCallback;
-import org.lealone.net.IntAsyncCallback;
 import org.lealone.net.Transfer;
-import org.lealone.net.VoidAsyncCallback;
 import org.lealone.storage.StorageCommand;
 
 /**
@@ -77,7 +75,7 @@ private void prepare(ClientSession s, boolean createParams) {
                 transfer.writeRequestHeader(id, Session.COMMAND_PREPARE);
             }
             transfer.writeInt(session.getSessionId()).writeString(sql);
-            VoidAsyncCallback ac = new VoidAsyncCallback() {
+            AsyncCallback<Void> ac = new AsyncCallback<Void>() {
                 @Override
                 public void runInternal() {
                     try {
@@ -250,6 +248,8 @@ public void runInternal() {
                 }
             }
         };
+        if (async)
+            ac.setAsyncHandler(handler);
         transfer.addAsyncCallback(id, ac);
         transfer.flush();
 
@@ -329,7 +329,7 @@ private int executeUpdate(String replicationName, AsyncHandler<AsyncResult<Integ
     private int getUpdateCount(boolean isDistributedUpdate, int id, AsyncHandler<AsyncResult<Integer>> handler,
             boolean async) throws IOException {
         isQuery = false;
-        IntAsyncCallback ac = new IntAsyncCallback() {
+        AsyncCallback<Integer> ac = new AsyncCallback<Integer>() {
             @Override
             public void runInternal() {
                 try {
@@ -348,14 +348,15 @@ public void runInternal() {
                 }
             }
         };
+        if (async)
+            ac.setAsyncHandler(handler);
         transfer.addAsyncCallback(id, ac);
         transfer.flush();
 
         int updateCount;
         if (async) {
             updateCount = -1;
         } else {
-            ac.await();
             updateCount = ac.getResult();
         }
 

lealone-client/src/main/java/org/lealone/client/ClientSession.java

@@ -37,8 +37,8 @@
 import org.lealone.db.SetTypes;
 import org.lealone.db.SysProperties;
 import org.lealone.db.value.Value;
+import org.lealone.net.AsyncCallback;
 import org.lealone.net.AsyncConnection;
-import org.lealone.net.OkAsyncCallback;
 import org.lealone.net.Transfer;
 import org.lealone.replication.ReplicationSession;
 import org.lealone.sql.ParsedStatement;
@@ -290,7 +290,7 @@ private void setAutoCommitSend(boolean autoCommit) {
             traceOperation("SESSION_SET_AUTOCOMMIT", autoCommit ? 1 : 0);
             transfer.writeRequestHeader(id, Session.SESSION_SET_AUTO_COMMIT);
             transfer.writeInt(sessionId).writeBoolean(autoCommit);
-            OkAsyncCallback ac = new OkAsyncCallback();
+            AsyncCallback<Void> ac = new AsyncCallback<>();
             transfer.addAsyncCallback(id, ac);
             transfer.flush();
             ac.await();

lealone-client/src/main/java/org/lealone/client/jdbc/JdbcStatement.java

@@ -93,15 +93,20 @@ private ResultSet executeQuery(String sql, AsyncHandler<AsyncResult<ResultSet>>
                 AsyncHandler<AsyncResult<Result>> h = new AsyncHandler<AsyncResult<Result>>() {
                     @Override
                     public void handle(AsyncResult<Result> ar) {
-                        Result r = ar.getResult();
-                        resultSet = new JdbcResultSet(conn, JdbcStatement.this, r, id, closedByResultSet, scrollable,
-                                updatable);
-                        resultSet.setCommand(command);
+                        if (ar.isSucceeded()) {
+                            Result r = ar.getResult();
+                            resultSet = new JdbcResultSet(conn, JdbcStatement.this, r, id, closedByResultSet,
+                                    scrollable, updatable);
+                            resultSet.setCommand(command);
+                        }
                         setExecutingStatement(null);
 
                         if (handler != null) {
                             AsyncResult<ResultSet> r2 = new AsyncResult<>();
-                            r2.setResult(resultSet);
+                            if (ar.isSucceeded())
+                                r2.setResult(resultSet);
+                            else
+                                r2.setCause(ar.getCause());
                             handler.handle(r2);
                         }
                     }
@@ -173,7 +178,8 @@ private int executeUpdateInternal(String sql, AsyncHandler<AsyncResult<Integer>>
             AsyncHandler<AsyncResult<Integer>> h = new AsyncHandler<AsyncResult<Integer>>() {
                 @Override
                 public void handle(AsyncResult<Integer> ar) {
-                    updateCount = ar.getResult();
+                    if (ar.isSucceeded())
+                        updateCount = ar.getResult();
                     // 设置完后再调用handle，否则有可能当前语句提前关闭了
                     setExecutingStatement(null);
                     command.close();

lealone-client/src/main/java/org/lealone/client/result/ClientResult.java

@@ -16,8 +16,8 @@
 import org.lealone.db.SysProperties;
 import org.lealone.db.result.Result;
 import org.lealone.db.value.Value;
+import org.lealone.net.AsyncCallback;
 import org.lealone.net.Transfer;
-import org.lealone.net.VoidAsyncCallback;
 
 /**
  * The client side part of a result set that is kept on the server.
@@ -168,7 +168,7 @@ protected void sendFetch(int fetchSize) throws IOException {
         session.traceOperation("RESULT_FETCH_ROWS", id);
         transfer.writeRequestHeader(id, Session.RESULT_FETCH_ROWS).writeInt(fetchSize);
 
-        VoidAsyncCallback ac = new VoidAsyncCallback();
+        AsyncCallback<Void> ac = new AsyncCallback<>();
         transfer.addAsyncCallback(id, ac);
         transfer.flush();
         ac.await();

lealone-common/src/main/java/org/lealone/async/AsyncResult.java

@@ -18,13 +18,46 @@
 package org.lealone.async;
 
 public class AsyncResult<T> {
+
     protected T result;
+    protected Throwable cause;
+    protected boolean succeeded;
+    protected boolean failed;
 
     public T getResult() {
         return result;
     }
 
     public void setResult(T result) {
         this.result = result;
+        failed = false;
+        succeeded = true;
+    }
+
+    public Throwable getCause() {
+        return cause;
+    }
+
+    public void setCause(Throwable cause) {
+        this.cause = cause;
+        failed = true;
+        succeeded = false;
+    }
+
+    public boolean isSucceeded() {
+        return succeeded;
     }
+
+    public void setSucceeded(boolean succeeded) {
+        this.succeeded = succeeded;
+    }
+
+    public boolean isFailed() {
+        return failed;
+    }
+
+    public void setFailed(boolean failed) {
+        this.failed = failed;
+    }
+
 }

lealone-net/src/main/java/org/lealone/net/AsyncCallback.java

@@ -19,20 +19,46 @@
 
 import java.util.concurrent.CountDownLatch;
 
+import org.lealone.async.AsyncHandler;
+import org.lealone.async.AsyncResult;
 import org.lealone.common.exceptions.DbException;
 
+@SuppressWarnings("rawtypes")
 public class AsyncCallback<T> {
 
     protected Transfer transfer;
     protected T result;
     protected DbException e;
     protected CountDownLatch latch = new CountDownLatch(1);
 
+    protected AsyncHandler ah;
+
+    public AsyncCallback() {
+    }
+
+    public AsyncCallback(AsyncHandler ah) {
+        this.ah = ah;
+    }
+
+    public void setAsyncHandler(AsyncHandler ah) {
+        this.ah = ah;
+    }
+
+    @SuppressWarnings("unchecked")
+    private void handle(DbException e) {
+        if (ah == null)
+            throw e;
+
+        AsyncResult r = new AsyncResult();
+        r.setCause(e);
+        ah.handle(r);
+    }
+
     public T getResult() {
         try {
             latch.await();
             if (e != null)
-                throw e;
+                handle(e);
         } catch (InterruptedException e) {
             throw DbException.convert(e);
         }
@@ -46,21 +72,18 @@ public void setTransfer(Transfer transfer) {
     public void setDbException(DbException e) {
         this.e = e;
         latch.countDown();
+        handle(e);
     }
 
     public void setResult(T result) {
         this.result = result;
         latch.countDown();
     }
 
-    public void run() {
-        runInternal();
-        latch.countDown();
-    }
-
     public void run(Transfer transfer) {
         this.transfer.setDataInputStream(transfer.getDataInputStream());
-        run();
+        runInternal();
+        latch.countDown();
     }
 
     protected void runInternal() {
@@ -70,7 +93,7 @@ public void await() {
         try {
             latch.await();
             if (e != null)
-                throw e;
+                handle(e);
         } catch (InterruptedException e) {
             throw DbException.convert(e);
         }

lealone-net/src/main/java/org/lealone/net/AsyncConnection.java

@@ -141,7 +141,7 @@ public void writeInitPacket(Session session, int sessionId, Transfer transfer, C
         for (String key : keys) {
             transfer.writeString(key).writeString(ci.getProperty(key));
         }
-        VoidAsyncCallback ac = new VoidAsyncCallback() {
+        AsyncCallback<Void> ac = new AsyncCallback<Void>() {
             @Override
             public void runInternal() {
                 try {
@@ -350,7 +350,7 @@ private static void writeBatchResult(Transfer transfer, Session session, int id,
         transfer.flush();
     }
 
-    private void executeQuery(Transfer transfer, Session session, int sessionId, int id, PreparedStatement command,
+    private void executeQueryAsync(Transfer transfer, Session session, int sessionId, int id, PreparedStatement command,
             int operation, int objectId, int maxRows, int fetchSize) throws IOException {
         PreparedCommand pc = new PreparedCommand(id, command, transfer, session, new Callable<Object>() {
             @Override
@@ -402,7 +402,7 @@ private void prepareCommit(Session session, PreparedStatement command) throws Ex
         }
     }
 
-    private void executeUpdate(Transfer transfer, Session session, int sessionId, int id, PreparedStatement command,
+    private void executeUpdateAsync(Transfer transfer, Session session, int sessionId, int id, PreparedStatement command,
             int operation) throws IOException {
         PreparedCommand pc = new PreparedCommand(id, command, transfer, session, new Callable<Object>() {
             @Override
@@ -469,7 +469,7 @@ void sendError(Transfer transfer, int id, Throwable t) {
             }
 
             if (isServer) {
-                message = "[Server]" + message;
+                message = "[Server] " + message;
             }
 
             transfer.reset(); // 为什么要reset? 见reset中的注释
@@ -573,7 +573,7 @@ private void processRequest(Transfer transfer, int id) throws IOException {
             }
             PreparedStatement command = session.prepareStatement(sql, fetchSize);
             cache.addObject(id, command);
-            executeQuery(transfer, session, sessionId, id, command, operation, objectId, maxRows, fetchSize);
+            executeQueryAsync(transfer, session, sessionId, id, command, operation, objectId, maxRows, fetchSize);
             break;
         }
         case Session.COMMAND_DISTRIBUTED_TRANSACTION_PREPARED_QUERY:
@@ -590,7 +590,7 @@ private void processRequest(Transfer transfer, int id) throws IOException {
                 session.setAutoCommit(false);
                 session.setRoot(false);
             }
-            executeQuery(transfer, session, sessionId, id, command, operation, objectId, maxRows, fetchSize);
+            executeQueryAsync(transfer, session, sessionId, id, command, operation, objectId, maxRows, fetchSize);
             break;
         }
         case Session.COMMAND_DISTRIBUTED_TRANSACTION_UPDATE:
@@ -607,7 +607,7 @@ private void processRequest(Transfer transfer, int id) throws IOException {
             }
             PreparedStatement command = session.prepareStatement(sql, -1);
             cache.addObject(id, command);
-            executeUpdate(transfer, session, sessionId, id, command, operation);
+            executeUpdateAsync(transfer, session, sessionId, id, command, operation);
             break;
         }
         case Session.COMMAND_DISTRIBUTED_TRANSACTION_PREPARED_UPDATE:
@@ -623,7 +623,7 @@ private void processRequest(Transfer transfer, int id) throws IOException {
                 session.setAutoCommit(false);
                 session.setRoot(false);
             }
-            executeUpdate(transfer, session, sessionId, id, command, operation);
+            executeUpdateAsync(transfer, session, sessionId, id, command, operation);
             break;
         }
         case Session.COMMAND_STORAGE_DISTRIBUTED_PUT: