cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject [1/5] git commit: inline IAsyncResult
Date Fri, 22 Mar 2013 15:58:14 GMT
Updated Branches:
  refs/heads/trunk 4ee8369b3 -> 292312380


inline IAsyncResult


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/e2ecb8ab
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/e2ecb8ab
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/e2ecb8ab

Branch: refs/heads/trunk
Commit: e2ecb8ab5863c1d1541f5be84a2bb7b67c5c9bb0
Parents: 4ee8369
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Fri Mar 22 10:37:41 2013 -0500
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Fri Mar 22 10:37:41 2013 -0500

----------------------------------------------------------------------
 src/java/org/apache/cassandra/net/AsyncResult.java |   10 +---
 .../org/apache/cassandra/net/IAsyncResult.java     |   42 ---------------
 .../org/apache/cassandra/net/MessagingService.java |    4 +-
 .../apache/cassandra/net/ResponseVerbHandler.java  |    2 +-
 .../service/RangeSliceResponseResolver.java        |    4 +-
 .../apache/cassandra/service/RowDataResolver.java  |   13 ++---
 .../apache/cassandra/service/StorageService.java   |    7 +--
 .../org/apache/cassandra/utils/FBUtilities.java    |    7 +--
 8 files changed, 15 insertions(+), 74 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/net/AsyncResult.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/AsyncResult.java b/src/java/org/apache/cassandra/net/AsyncResult.java
index bb166b6..ebeda46 100644
--- a/src/java/org/apache/cassandra/net/AsyncResult.java
+++ b/src/java/org/apache/cassandra/net/AsyncResult.java
@@ -17,7 +17,6 @@
  */
 package org.apache.cassandra.net;
 
-import java.net.InetAddress;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -25,14 +24,13 @@ import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-class AsyncResult<T> implements IAsyncResult<T>
+public class AsyncResult<T> implements IMessageCallback
 {
     private T result;
     private final AtomicBoolean done = new AtomicBoolean(false);
     private final Lock lock = new ReentrantLock();
     private final Condition condition;
     private final long startTime;
-    private InetAddress from;
 
     public AsyncResult()
     {
@@ -79,7 +77,6 @@ class AsyncResult<T> implements IAsyncResult<T>
             lock.lock();
             if (!done.get())
             {
-                from = response.from;
                 result = response.payload;
                 done.set(true);
                 condition.signal();
@@ -95,9 +92,4 @@ class AsyncResult<T> implements IAsyncResult<T>
     {
         return false;
     }
-
-    public InetAddress getFrom()
-    {
-        return from;
-    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/net/IAsyncResult.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/IAsyncResult.java b/src/java/org/apache/cassandra/net/IAsyncResult.java
deleted file mode 100644
index 87a4c73..0000000
--- a/src/java/org/apache/cassandra/net/IAsyncResult.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.net;
-
-import java.net.InetAddress;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-public interface IAsyncResult<T> extends IMessageCallback
-{
-    /**
-     * Same operation as the above get() but allows the calling
-     * thread to specify a timeout.
-     * @param timeout the maximum time to wait
-     * @param tu the time unit of the timeout argument
-     * @return the result wrapped in an Object[]
-    */
-    public T get(long timeout, TimeUnit tu) throws TimeoutException;
-
-    /**
-     * Store the result obtained for the submitted task.
-     * @param result the response message
-     */
-    public void result(MessageIn<T> result);
-
-    public InetAddress getFrom();
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index 4fbaffe..789555d 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -615,9 +615,9 @@ public final class MessagingService implements MessagingServiceMBean
         connection.enqueue(processedMessage, id);
     }
 
-    public <T> IAsyncResult<T> sendRR(MessageOut message, InetAddress to)
+    public <T> AsyncResult<T> sendRR(MessageOut message, InetAddress to)
     {
-        IAsyncResult<T> iar = new AsyncResult<T>();
+        AsyncResult<T> iar = new AsyncResult<T>();
         sendRR(message, to, iar);
         return iar;
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
index b3a5f32..92fbad1 100644
--- a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
+++ b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java
@@ -49,7 +49,7 @@ public class ResponseVerbHandler implements IVerbHandler
         else
         {
             Tracing.trace("Processing result from {}", message.from);
-            ((IAsyncResult) cb).result(message);
+            ((AsyncResult) cb).result(message);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
index 2049954..e14dd26 100644
--- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
+++ b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
@@ -24,7 +24,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import com.google.common.collect.AbstractIterator;
 
 import org.apache.cassandra.db.*;
-import org.apache.cassandra.net.IAsyncResult;
+import org.apache.cassandra.net.AsyncResult;
 import org.apache.cassandra.net.MessageIn;
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.CloseableIterator;
@@ -47,7 +47,7 @@ public class RangeSliceResponseResolver implements IResponseResolver<RangeSliceR
     private final String table;
     private List<InetAddress> sources;
     protected final Collection<MessageIn<RangeSliceReply>> responses = new LinkedBlockingQueue<MessageIn<RangeSliceReply>>();;
-    public final List<IAsyncResult> repairResults = new ArrayList<IAsyncResult>();
+    public final List<AsyncResult> repairResults = new ArrayList<AsyncResult>();
 
     public RangeSliceResponseResolver(String table)
     {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/service/RowDataResolver.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/RowDataResolver.java b/src/java/org/apache/cassandra/service/RowDataResolver.java
index 1d57531..f992232 100644
--- a/src/java/org/apache/cassandra/service/RowDataResolver.java
+++ b/src/java/org/apache/cassandra/service/RowDataResolver.java
@@ -30,19 +30,14 @@ import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
 import org.apache.cassandra.db.filter.IDiskAtomFilter;
 import org.apache.cassandra.db.filter.QueryFilter;
-import org.apache.cassandra.db.filter.SliceQueryFilter;
-import org.apache.cassandra.net.IAsyncResult;
-import org.apache.cassandra.net.MessageIn;
-import org.apache.cassandra.net.MessageOut;
-import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.net.*;
 import org.apache.cassandra.utils.CloseableIterator;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.IFilter;
 
 public class RowDataResolver extends AbstractRowResolver
 {
     private int maxLiveCount = 0;
-    public List<IAsyncResult> repairResults = Collections.emptyList();
+    public List<AsyncResult> repairResults = Collections.emptyList();
     private final IDiskAtomFilter filter;
 
     public RowDataResolver(String table, ByteBuffer key, IDiskAtomFilter qFilter)
@@ -108,9 +103,9 @@ public class RowDataResolver extends AbstractRowResolver
      * For each row version, compare with resolved (the superset of all row versions);
      * if it is missing anything, send a mutation to the endpoint it come from.
      */
-    public static List<IAsyncResult> scheduleRepairs(ColumnFamily resolved, String
table, DecoratedKey key, List<ColumnFamily> versions, List<InetAddress> endpoints)
+    public static List<AsyncResult> scheduleRepairs(ColumnFamily resolved, String table,
DecoratedKey key, List<ColumnFamily> versions, List<InetAddress> endpoints)
     {
-        List<IAsyncResult> results = new ArrayList<IAsyncResult>(versions.size());
+        List<AsyncResult> results = new ArrayList<AsyncResult>(versions.size());
 
         for (int i = 0; i < versions.size(); i++)
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/service/StorageService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 3e08505..803fad8 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -65,10 +65,7 @@ import org.apache.cassandra.io.sstable.SSTableLoader;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.locator.*;
 import org.apache.cassandra.metrics.StorageMetrics;
-import org.apache.cassandra.net.IAsyncResult;
-import org.apache.cassandra.net.MessageOut;
-import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.net.ResponseVerbHandler;
+import org.apache.cassandra.net.*;
 import org.apache.cassandra.service.AntiEntropyService.TreeRequestVerbHandler;
 import org.apache.cassandra.streaming.*;
 import org.apache.cassandra.thrift.cassandraConstants;
@@ -1785,7 +1782,7 @@ public class StorageService extends NotificationBroadcasterSupport implements
IE
             logger.debug("Notifying " + remote.toString() + " of replication completion\n");
         while (failureDetector.isAlive(remote))
         {
-            IAsyncResult iar = MessagingService.instance().sendRR(msg, remote);
+            AsyncResult iar = MessagingService.instance().sendRR(msg, remote);
             try
             {
                 iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/e2ecb8ab/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index e6e2e50..d6e4dbd 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -19,7 +19,6 @@ package org.apache.cassandra.utils;
 
 import java.io.*;
 import java.lang.reflect.Field;
-import java.lang.reflect.InvocationTargetException;
 import java.math.BigInteger;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
@@ -53,7 +52,7 @@ import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.io.util.IAllocator;
-import org.apache.cassandra.net.IAsyncResult;
+import org.apache.cassandra.net.AsyncResult;
 import org.apache.thrift.TBase;
 import org.apache.thrift.TDeserializer;
 import org.apache.thrift.TException;
@@ -384,9 +383,9 @@ public class FBUtilities
         }
     }
 
-    public static void waitOnFutures(List<IAsyncResult> results, long ms) throws TimeoutException
+    public static void waitOnFutures(List<AsyncResult> results, long ms) throws TimeoutException
     {
-        for (IAsyncResult result : results)
+        for (AsyncResult result : results)
             result.get(ms, TimeUnit.MILLISECONDS);
     }
 


Mime
View raw message