cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r799331 [19/29] - in /incubator/cassandra/trunk: ./ src/java/org/apache/cassandra/concurrent/ src/java/org/apache/cassandra/config/ src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/dht/ src/java/org/apache/cassandra/gms/ src/...
Date Thu, 30 Jul 2009 15:30:27 GMT
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/TokenMetadata.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/TokenMetadata.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/TokenMetadata.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/locator/TokenMetadata.java Thu Jul 30 15:30:21 2009
@@ -1,173 +1,173 @@
-/**
- * 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.locator;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.net.EndPoint;
-
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public class TokenMetadata
-{
-    /* Maintains token to endpoint map of every node in the cluster. */
-    private Map<Token, EndPoint> tokenToEndPointMap_ = new HashMap<Token, EndPoint>();
-    /* Maintains a reverse index of endpoint to token in the cluster. */
-    private Map<EndPoint, Token> endPointToTokenMap_ = new HashMap<EndPoint, Token>();
-    
-    /* Use this lock for manipulating the token map */
-    private final ReadWriteLock lock_ = new ReentrantReadWriteLock(true);
-
-    public TokenMetadata()
-    {
-    }
-
-    private TokenMetadata(Map<Token, EndPoint> tokenToEndPointMap, Map<EndPoint, Token> endPointToTokenMap)
-    {
-        tokenToEndPointMap_ = tokenToEndPointMap;
-        endPointToTokenMap_ = endPointToTokenMap;
-    }
-    
-    public TokenMetadata cloneMe()
-    {
-        return new TokenMetadata(cloneTokenEndPointMap(), cloneEndPointTokenMap());
-    }
-    
-    /**
-     * Update the two maps in an safe mode. 
-    */
-    public void update(Token token, EndPoint endpoint)
-    {
-        lock_.writeLock().lock();
-        try
-        {            
-            Token oldToken = endPointToTokenMap_.get(endpoint);
-            if ( oldToken != null )
-                tokenToEndPointMap_.remove(oldToken);
-            tokenToEndPointMap_.put(token, endpoint);
-            endPointToTokenMap_.put(endpoint, token);
-        }
-        finally
-        {
-            lock_.writeLock().unlock();
-        }
-    }
-    
-    /**
-     * Remove the entries in the two maps.
-     * @param endpoint
-     */
-    public void remove(EndPoint endpoint)
-    {
-        lock_.writeLock().lock();
-        try
-        {            
-            Token oldToken = endPointToTokenMap_.get(endpoint);
-            if ( oldToken != null )
-                tokenToEndPointMap_.remove(oldToken);            
-            endPointToTokenMap_.remove(endpoint);
-        }
-        finally
-        {
-            lock_.writeLock().unlock();
-        }
-    }
-    
-    public Token getToken(EndPoint endpoint)
-    {
-        lock_.readLock().lock();
-        try
-        {
-            return endPointToTokenMap_.get(endpoint);
-        }
-        finally
-        {
-            lock_.readLock().unlock();
-        }
-    }
-    
-    public boolean isKnownEndPoint(EndPoint ep)
-    {
-        lock_.readLock().lock();
-        try
-        {
-            return endPointToTokenMap_.containsKey(ep);
-        }
-        finally
-        {
-            lock_.readLock().unlock();
-        }
-    }
-    
-    /*
-     * Returns a safe clone of tokenToEndPointMap_.
-    */
-    public Map<Token, EndPoint> cloneTokenEndPointMap()
-    {
-        lock_.readLock().lock();
-        try
-        {            
-            return new HashMap<Token, EndPoint>( tokenToEndPointMap_ );
-        }
-        finally
-        {
-            lock_.readLock().unlock();
-        }
-    }
-    
-    /*
-     * Returns a safe clone of endPointTokenMap_.
-    */
-    public Map<EndPoint, Token> cloneEndPointTokenMap()
-    {
-        lock_.readLock().lock();
-        try
-        {            
-            return new HashMap<EndPoint, Token>( endPointToTokenMap_ );
-        }
-        finally
-        {
-            lock_.readLock().unlock();
-        }
-    }
-    
-    public String toString()
-    {
-        StringBuilder sb = new StringBuilder();
-        Set<EndPoint> eps = endPointToTokenMap_.keySet();
-        
-        for ( EndPoint ep : eps )
-        {
-            sb.append(ep);
-            sb.append(":");
-            sb.append(endPointToTokenMap_.get(ep));
-            sb.append(System.getProperty("line.separator"));
-        }
-        
-        return sb.toString();
-    }
-}
+/**
+ * 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.locator;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.net.EndPoint;
+
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class TokenMetadata
+{
+    /* Maintains token to endpoint map of every node in the cluster. */
+    private Map<Token, EndPoint> tokenToEndPointMap_ = new HashMap<Token, EndPoint>();
+    /* Maintains a reverse index of endpoint to token in the cluster. */
+    private Map<EndPoint, Token> endPointToTokenMap_ = new HashMap<EndPoint, Token>();
+    
+    /* Use this lock for manipulating the token map */
+    private final ReadWriteLock lock_ = new ReentrantReadWriteLock(true);
+
+    public TokenMetadata()
+    {
+    }
+
+    private TokenMetadata(Map<Token, EndPoint> tokenToEndPointMap, Map<EndPoint, Token> endPointToTokenMap)
+    {
+        tokenToEndPointMap_ = tokenToEndPointMap;
+        endPointToTokenMap_ = endPointToTokenMap;
+    }
+    
+    public TokenMetadata cloneMe()
+    {
+        return new TokenMetadata(cloneTokenEndPointMap(), cloneEndPointTokenMap());
+    }
+    
+    /**
+     * Update the two maps in an safe mode. 
+    */
+    public void update(Token token, EndPoint endpoint)
+    {
+        lock_.writeLock().lock();
+        try
+        {            
+            Token oldToken = endPointToTokenMap_.get(endpoint);
+            if ( oldToken != null )
+                tokenToEndPointMap_.remove(oldToken);
+            tokenToEndPointMap_.put(token, endpoint);
+            endPointToTokenMap_.put(endpoint, token);
+        }
+        finally
+        {
+            lock_.writeLock().unlock();
+        }
+    }
+    
+    /**
+     * Remove the entries in the two maps.
+     * @param endpoint
+     */
+    public void remove(EndPoint endpoint)
+    {
+        lock_.writeLock().lock();
+        try
+        {            
+            Token oldToken = endPointToTokenMap_.get(endpoint);
+            if ( oldToken != null )
+                tokenToEndPointMap_.remove(oldToken);            
+            endPointToTokenMap_.remove(endpoint);
+        }
+        finally
+        {
+            lock_.writeLock().unlock();
+        }
+    }
+    
+    public Token getToken(EndPoint endpoint)
+    {
+        lock_.readLock().lock();
+        try
+        {
+            return endPointToTokenMap_.get(endpoint);
+        }
+        finally
+        {
+            lock_.readLock().unlock();
+        }
+    }
+    
+    public boolean isKnownEndPoint(EndPoint ep)
+    {
+        lock_.readLock().lock();
+        try
+        {
+            return endPointToTokenMap_.containsKey(ep);
+        }
+        finally
+        {
+            lock_.readLock().unlock();
+        }
+    }
+    
+    /*
+     * Returns a safe clone of tokenToEndPointMap_.
+    */
+    public Map<Token, EndPoint> cloneTokenEndPointMap()
+    {
+        lock_.readLock().lock();
+        try
+        {            
+            return new HashMap<Token, EndPoint>( tokenToEndPointMap_ );
+        }
+        finally
+        {
+            lock_.readLock().unlock();
+        }
+    }
+    
+    /*
+     * Returns a safe clone of endPointTokenMap_.
+    */
+    public Map<EndPoint, Token> cloneEndPointTokenMap()
+    {
+        lock_.readLock().lock();
+        try
+        {            
+            return new HashMap<EndPoint, Token>( endPointToTokenMap_ );
+        }
+        finally
+        {
+            lock_.readLock().unlock();
+        }
+    }
+    
+    public String toString()
+    {
+        StringBuilder sb = new StringBuilder();
+        Set<EndPoint> eps = endPointToTokenMap_.keySet();
+        
+        for ( EndPoint ep : eps )
+        {
+            sb.append(ep);
+            sb.append(":");
+            sb.append(endPointToTokenMap_.get(ep));
+            sb.append(System.getProperty("line.separator"));
+        }
+        
+        return sb.toString();
+    }
+}

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/AsyncResult.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/AsyncResult.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/AsyncResult.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/AsyncResult.java Thu Jul 30 15:30:21 2009
@@ -1,132 +1,132 @@
-/**
- * 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.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.cassandra.utils.LogUtil;
-import org.apache.log4j.Logger;
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-class AsyncResult implements IAsyncResult
-{
-    private static Logger logger_ = Logger.getLogger( AsyncResult.class );
-    private byte[] result_;
-    private AtomicBoolean done_ = new AtomicBoolean(false);
-    private Lock lock_ = new ReentrantLock();
-    private Condition condition_;
-
-    public AsyncResult()
-    {        
-        condition_ = lock_.newCondition();
-    }    
-    
-    public byte[] get()
-    {
-        lock_.lock();
-        try
-        {
-            if ( !done_.get() )
-            {
-                condition_.await();                    
-            }
-        }
-        catch ( InterruptedException ex )
-        {
-            logger_.warn( LogUtil.throwableToString(ex) );
-        }
-        finally
-        {
-            lock_.unlock();            
-        }        
-        return result_;
-    }
-    
-    public boolean isDone()
-    {
-        return done_.get();
-    }
-    
-    public byte[] get(long timeout, TimeUnit tu) throws TimeoutException
-    {
-        lock_.lock();
-        try
-        {            
-            boolean bVal = true;
-            try
-            {
-                if ( !done_.get() )
-                {                    
-                    bVal = condition_.await(timeout, tu);
-                }
-            }
-            catch ( InterruptedException ex )
-            {
-                logger_.warn( LogUtil.throwableToString(ex) );
-            }
-            
-            if ( !bVal && !done_.get() )
-            {                                           
-                throw new TimeoutException("Operation timed out.");
-            }
-        }
-        finally
-        {
-            lock_.unlock();      
-        }
-        return result_;
-    }
-    
-    public List<byte[]> multiget()
-    {
-        throw new UnsupportedOperationException("This operation is not supported in the AsyncResult abstraction.");
-    }
-    
-    public List<byte[]> multiget(long timeout, TimeUnit tu) throws TimeoutException
-    {
-        throw new UnsupportedOperationException("This operation is not supported in the AsyncResult abstraction.");
-    }
-    
-    public void result(Message response)
-    {        
-        try
-        {
-            lock_.lock();
-            if ( !done_.get() )
-            {                
-                result_ = response.getMessageBody();
-                done_.set(true);
-                condition_.signal();
-            }
-        }
-        finally
-        {
-            lock_.unlock();
-        }        
-    }    
-}
+/**
+ * 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.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.cassandra.utils.LogUtil;
+import org.apache.log4j.Logger;
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+class AsyncResult implements IAsyncResult
+{
+    private static Logger logger_ = Logger.getLogger( AsyncResult.class );
+    private byte[] result_;
+    private AtomicBoolean done_ = new AtomicBoolean(false);
+    private Lock lock_ = new ReentrantLock();
+    private Condition condition_;
+
+    public AsyncResult()
+    {        
+        condition_ = lock_.newCondition();
+    }    
+    
+    public byte[] get()
+    {
+        lock_.lock();
+        try
+        {
+            if ( !done_.get() )
+            {
+                condition_.await();                    
+            }
+        }
+        catch ( InterruptedException ex )
+        {
+            logger_.warn( LogUtil.throwableToString(ex) );
+        }
+        finally
+        {
+            lock_.unlock();            
+        }        
+        return result_;
+    }
+    
+    public boolean isDone()
+    {
+        return done_.get();
+    }
+    
+    public byte[] get(long timeout, TimeUnit tu) throws TimeoutException
+    {
+        lock_.lock();
+        try
+        {            
+            boolean bVal = true;
+            try
+            {
+                if ( !done_.get() )
+                {                    
+                    bVal = condition_.await(timeout, tu);
+                }
+            }
+            catch ( InterruptedException ex )
+            {
+                logger_.warn( LogUtil.throwableToString(ex) );
+            }
+            
+            if ( !bVal && !done_.get() )
+            {                                           
+                throw new TimeoutException("Operation timed out.");
+            }
+        }
+        finally
+        {
+            lock_.unlock();      
+        }
+        return result_;
+    }
+    
+    public List<byte[]> multiget()
+    {
+        throw new UnsupportedOperationException("This operation is not supported in the AsyncResult abstraction.");
+    }
+    
+    public List<byte[]> multiget(long timeout, TimeUnit tu) throws TimeoutException
+    {
+        throw new UnsupportedOperationException("This operation is not supported in the AsyncResult abstraction.");
+    }
+    
+    public void result(Message response)
+    {        
+        try
+        {
+            lock_.lock();
+            if ( !done_.get() )
+            {                
+                result_ = response.getMessageBody();
+                done_.set(true);
+                condition_.signal();
+            }
+        }
+        finally
+        {
+            lock_.unlock();
+        }        
+    }    
+}

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndPointSerializationHelper.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndPointSerializationHelper.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndPointSerializationHelper.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/CompactEndPointSerializationHelper.java Thu Jul 30 15:30:21 2009
@@ -1,50 +1,50 @@
-/**
- * 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.io.*;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public class CompactEndPointSerializationHelper
-{
-    public static void serialize(EndPoint endPoint, DataOutputStream dos) throws IOException
-    {        
-        dos.write(EndPoint.toBytes(endPoint));
-    }
-    
-    public static EndPoint deserialize(DataInputStream dis) throws IOException
-    {     
-        byte[] bytes = new byte[6];
-        dis.readFully(bytes, 0, bytes.length);
-        return EndPoint.fromBytes(bytes);       
-    }
-    
-    public static void main(String[] args) throws Throwable
-    {
-        EndPoint ep = new EndPoint(7000);
-        byte[] bytes = EndPoint.toBytes(ep);
-        System.out.println(bytes.length);
-        EndPoint ep2 = EndPoint.fromBytes(bytes);
-        System.out.println(ep2);
-    }
+/**
+ * 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.io.*;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class CompactEndPointSerializationHelper
+{
+    public static void serialize(EndPoint endPoint, DataOutputStream dos) throws IOException
+    {        
+        dos.write(EndPoint.toBytes(endPoint));
+    }
+    
+    public static EndPoint deserialize(DataInputStream dis) throws IOException
+    {     
+        byte[] bytes = new byte[6];
+        dis.readFully(bytes, 0, bytes.length);
+        return EndPoint.fromBytes(bytes);       
+    }
+    
+    public static void main(String[] args) throws Throwable
+    {
+        EndPoint ep = new EndPoint(7000);
+        byte[] bytes = EndPoint.toBytes(ep);
+        System.out.println(bytes.length);
+        EndPoint ep2 = EndPoint.fromBytes(bytes);
+        System.out.println(ep2);
+    }
 }
\ No newline at end of file

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/ConnectionStatistics.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/ConnectionStatistics.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/ConnectionStatistics.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/ConnectionStatistics.java Thu Jul 30 15:30:21 2009
@@ -1,78 +1,78 @@
-/**
- * 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;
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public class ConnectionStatistics
-{
-    private String localHost;
-    private int localPort;
-    private String remoteHost;
-    private int remotePort;
-    private int totalConnections;
-    private int connectionsInUse;
-
-    ConnectionStatistics(EndPoint localEp, EndPoint remoteEp, int tc, int ciu)
-    {
-        localHost = localEp.getHost();
-        localPort = localEp.getPort();
-        remoteHost = remoteEp.getHost();
-        remotePort = remoteEp.getPort();
-        totalConnections = tc;
-        connectionsInUse = ciu;
-    }
-    
-    public String getLocalHost()
-    {
-        return localHost;
-    }
-    
-    public int getLocalPort()
-    {
-        return localPort;
-    }
-    
-    public String getRemoteHost()
-    {
-        return remoteHost;
-    }
-    
-    public int getRemotePort()
-    {
-        return remotePort;
-    }
-    
-    public int getTotalConnections()
-    {
-        return totalConnections;
-    }
-    
-    public int getConnectionInUse()
-    {
-        return connectionsInUse;
-    }
-
-    public String toString()
-    {
-        return localHost + ":" + localPort + "->" + remoteHost + ":" + remotePort + " Total Connections open : " + totalConnections + " Connections in use : " + connectionsInUse;
-    }
+/**
+ * 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;
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class ConnectionStatistics
+{
+    private String localHost;
+    private int localPort;
+    private String remoteHost;
+    private int remotePort;
+    private int totalConnections;
+    private int connectionsInUse;
+
+    ConnectionStatistics(EndPoint localEp, EndPoint remoteEp, int tc, int ciu)
+    {
+        localHost = localEp.getHost();
+        localPort = localEp.getPort();
+        remoteHost = remoteEp.getHost();
+        remotePort = remoteEp.getPort();
+        totalConnections = tc;
+        connectionsInUse = ciu;
+    }
+    
+    public String getLocalHost()
+    {
+        return localHost;
+    }
+    
+    public int getLocalPort()
+    {
+        return localPort;
+    }
+    
+    public String getRemoteHost()
+    {
+        return remoteHost;
+    }
+    
+    public int getRemotePort()
+    {
+        return remotePort;
+    }
+    
+    public int getTotalConnections()
+    {
+        return totalConnections;
+    }
+    
+    public int getConnectionInUse()
+    {
+        return connectionsInUse;
+    }
+
+    public String toString()
+    {
+        return localHost + ":" + localPort + "->" + remoteHost + ":" + remotePort + " Total Connections open : " + totalConnections + " Connections in use : " + connectionsInUse;
+    }
 }
\ No newline at end of file

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/EndPoint.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/EndPoint.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/EndPoint.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/EndPoint.java Thu Jul 30 15:30:21 2009
@@ -1,175 +1,175 @@
-/**
- * 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.io.IOException;
-import java.io.Serializable;
-import java.net.*;
-import java.nio.ByteBuffer;
-import java.nio.CharBuffer;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.LogUtil;
-import org.apache.log4j.Logger;
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public class EndPoint implements Serializable, Comparable<EndPoint>
-{
-    // logging and profiling.
-    private static Logger logger_ = Logger.getLogger(EndPoint.class);
-    private static final long serialVersionUID = -4962625949179835907L;
-    private static Map<CharBuffer, String> hostNames_ = new HashMap<CharBuffer, String>();
-
-    // use as a kind of magic number to send ourselves a message indicating listening state
-    protected static final int sentinelPort_ = 5555;
-    public static EndPoint sentinelLocalEndPoint_;
-    
-    static
-    {
-        try
-        {
-            sentinelLocalEndPoint_ = new EndPoint(FBUtilities.getHostAddress(), EndPoint.sentinelPort_);
-        }        
-        catch ( IOException ex )
-        {
-            logger_.warn(LogUtil.throwableToString(ex));
-        }
-    }
-
-    private String host_;
-    private int port_;
-
-    private transient InetSocketAddress ia_;
-
-    public EndPoint(String host, int port)
-    {
-        assert host.matches("\\d+\\.\\d+\\.\\d+\\.\\d+") : host;
-        host_ = host;
-        port_ = port;
-    }
-
-    // create a local endpoint id
-    public EndPoint(int port)
-    {
-        try
-        {
-            host_ = FBUtilities.getHostAddress();
-        }
-        catch (UnknownHostException e)
-        {
-            throw new RuntimeException(e);
-        }
-        port_ = port;
-    }
-
-    public String getHost()
-    {
-        return host_;
-    }
-
-    public int getPort()
-    {
-        return port_;
-    }
-
-    public void setPort(int port)
-    {
-        port_ = port;
-    }
-
-    public InetSocketAddress getInetAddress()
-    {
-        if (ia_ == null || ia_.isUnresolved())
-        {
-            ia_ = new InetSocketAddress(host_, port_);
-        }
-        return ia_;
-    }
-
-    public boolean equals(Object o)
-    {
-        if (!(o instanceof EndPoint))
-            return false;
-
-        EndPoint rhs = (EndPoint) o;
-        return (host_.equals(rhs.host_) && port_ == rhs.port_);
-    }
-
-    public int hashCode()
-    {
-        return (host_ + port_).hashCode();
-    }
-
-    public int compareTo(EndPoint rhs)
-    {
-        return host_.compareTo(rhs.host_);
-    }
-
-    public String toString()
-    {
-        return (host_ + ":" + port_);
-    }
-
-    public static EndPoint fromString(String str)
-    {
-        String[] values = str.split(":");
-        return new EndPoint(values[0], Integer.parseInt(values[1]));
-    }
-
-    public static byte[] toBytes(EndPoint ep)
-    {
-        ByteBuffer buffer = ByteBuffer.allocate(6);
-        byte[] iaBytes = ep.getInetAddress().getAddress().getAddress();
-        buffer.put(iaBytes);
-        buffer.put(MessagingService.toByteArray((short) ep.getPort()));
-        buffer.flip();
-        return buffer.array();
-    }
-
-    public static EndPoint fromBytes(byte[] bytes)
-    {
-        ByteBuffer buffer = ByteBuffer.allocate(4);
-        System.arraycopy(bytes, 0, buffer.array(), 0, 4);
-        byte[] portBytes = new byte[2];
-        System.arraycopy(bytes, 4, portBytes, 0, portBytes.length);
-        try
-        {
-            CharBuffer charBuffer = buffer.asCharBuffer();
-            String host = hostNames_.get(charBuffer);
-            if (host == null)
-            {               
-                host = InetAddress.getByAddress(buffer.array()).getHostAddress();              
-                hostNames_.put(charBuffer, host);
-            }
-            int port = (int) MessagingService.byteArrayToShort(portBytes);
-            return new EndPoint(host, port);
-        }
-        catch (UnknownHostException e)
-        {
-            throw new IllegalArgumentException(e);
-        }
-    }
-}
-
+/**
+ * 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.io.IOException;
+import java.io.Serializable;
+import java.net.*;
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.LogUtil;
+import org.apache.log4j.Logger;
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class EndPoint implements Serializable, Comparable<EndPoint>
+{
+    // logging and profiling.
+    private static Logger logger_ = Logger.getLogger(EndPoint.class);
+    private static final long serialVersionUID = -4962625949179835907L;
+    private static Map<CharBuffer, String> hostNames_ = new HashMap<CharBuffer, String>();
+
+    // use as a kind of magic number to send ourselves a message indicating listening state
+    protected static final int sentinelPort_ = 5555;
+    public static EndPoint sentinelLocalEndPoint_;
+    
+    static
+    {
+        try
+        {
+            sentinelLocalEndPoint_ = new EndPoint(FBUtilities.getHostAddress(), EndPoint.sentinelPort_);
+        }        
+        catch ( IOException ex )
+        {
+            logger_.warn(LogUtil.throwableToString(ex));
+        }
+    }
+
+    private String host_;
+    private int port_;
+
+    private transient InetSocketAddress ia_;
+
+    public EndPoint(String host, int port)
+    {
+        assert host.matches("\\d+\\.\\d+\\.\\d+\\.\\d+") : host;
+        host_ = host;
+        port_ = port;
+    }
+
+    // create a local endpoint id
+    public EndPoint(int port)
+    {
+        try
+        {
+            host_ = FBUtilities.getHostAddress();
+        }
+        catch (UnknownHostException e)
+        {
+            throw new RuntimeException(e);
+        }
+        port_ = port;
+    }
+
+    public String getHost()
+    {
+        return host_;
+    }
+
+    public int getPort()
+    {
+        return port_;
+    }
+
+    public void setPort(int port)
+    {
+        port_ = port;
+    }
+
+    public InetSocketAddress getInetAddress()
+    {
+        if (ia_ == null || ia_.isUnresolved())
+        {
+            ia_ = new InetSocketAddress(host_, port_);
+        }
+        return ia_;
+    }
+
+    public boolean equals(Object o)
+    {
+        if (!(o instanceof EndPoint))
+            return false;
+
+        EndPoint rhs = (EndPoint) o;
+        return (host_.equals(rhs.host_) && port_ == rhs.port_);
+    }
+
+    public int hashCode()
+    {
+        return (host_ + port_).hashCode();
+    }
+
+    public int compareTo(EndPoint rhs)
+    {
+        return host_.compareTo(rhs.host_);
+    }
+
+    public String toString()
+    {
+        return (host_ + ":" + port_);
+    }
+
+    public static EndPoint fromString(String str)
+    {
+        String[] values = str.split(":");
+        return new EndPoint(values[0], Integer.parseInt(values[1]));
+    }
+
+    public static byte[] toBytes(EndPoint ep)
+    {
+        ByteBuffer buffer = ByteBuffer.allocate(6);
+        byte[] iaBytes = ep.getInetAddress().getAddress().getAddress();
+        buffer.put(iaBytes);
+        buffer.put(MessagingService.toByteArray((short) ep.getPort()));
+        buffer.flip();
+        return buffer.array();
+    }
+
+    public static EndPoint fromBytes(byte[] bytes)
+    {
+        ByteBuffer buffer = ByteBuffer.allocate(4);
+        System.arraycopy(bytes, 0, buffer.array(), 0, 4);
+        byte[] portBytes = new byte[2];
+        System.arraycopy(bytes, 4, portBytes, 0, portBytes.length);
+        try
+        {
+            CharBuffer charBuffer = buffer.asCharBuffer();
+            String host = hostNames_.get(charBuffer);
+            if (host == null)
+            {               
+                host = InetAddress.getByAddress(buffer.array()).getHostAddress();              
+                hostNames_.put(charBuffer, host);
+            }
+            int port = (int) MessagingService.byteArrayToShort(portBytes);
+            return new EndPoint(host, port);
+        }
+        catch (UnknownHostException e)
+        {
+            throw new IllegalArgumentException(e);
+        }
+    }
+}
+

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/FileStreamTask.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/FileStreamTask.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/FileStreamTask.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/FileStreamTask.java Thu Jul 30 15:30:21 2009
@@ -1,85 +1,85 @@
-/**
- * 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.io.*;
-import java.net.SocketException;
-
-import org.apache.cassandra.net.sink.SinkManager;
-import org.apache.cassandra.utils.LogUtil;
-import org.apache.log4j.Logger;
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-class FileStreamTask implements Runnable
-{
-    private static Logger logger_ = Logger.getLogger( FileStreamTask.class );
-    
-    private String file_;
-    private long startPosition_;
-    private long total_;
-    private EndPoint from_;
-    private EndPoint to_;
-    
-    FileStreamTask(String file, long startPosition, long total, EndPoint from, EndPoint to)
-    {
-        file_ = file;
-        startPosition_ = startPosition;
-        total_ = total;
-        from_ = from;
-        to_ = to;
-    }
-    
-    public void run()
-    {
-        TcpConnection connection = null;
-        try
-        {                        
-            connection = new TcpConnection(from_, to_);
-            File file = new File(file_);             
-            connection.stream(file, startPosition_, total_);
-            MessagingService.setStreamingMode(false);
-            if (logger_.isDebugEnabled())
-              logger_.debug("Done streaming " + file);
-        }            
-        catch ( SocketException se )
-        {                        
-            logger_.info(LogUtil.throwableToString(se));
-        }
-        catch ( IOException e )
-        {
-            logConnectAndIOException(e, connection);
-        }
-        catch (Throwable th)
-        {
-            logger_.warn(LogUtil.throwableToString(th));
-        }        
-    }
-    
-    private void logConnectAndIOException(IOException ex, TcpConnection connection)
-    {                    
-        if ( connection != null )
-        {
-            connection.errorClose();
-        }
-        logger_.info(LogUtil.throwableToString(ex));
-    }
-}
+/**
+ * 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.io.*;
+import java.net.SocketException;
+
+import org.apache.cassandra.net.sink.SinkManager;
+import org.apache.cassandra.utils.LogUtil;
+import org.apache.log4j.Logger;
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+class FileStreamTask implements Runnable
+{
+    private static Logger logger_ = Logger.getLogger( FileStreamTask.class );
+    
+    private String file_;
+    private long startPosition_;
+    private long total_;
+    private EndPoint from_;
+    private EndPoint to_;
+    
+    FileStreamTask(String file, long startPosition, long total, EndPoint from, EndPoint to)
+    {
+        file_ = file;
+        startPosition_ = startPosition;
+        total_ = total;
+        from_ = from;
+        to_ = to;
+    }
+    
+    public void run()
+    {
+        TcpConnection connection = null;
+        try
+        {                        
+            connection = new TcpConnection(from_, to_);
+            File file = new File(file_);             
+            connection.stream(file, startPosition_, total_);
+            MessagingService.setStreamingMode(false);
+            if (logger_.isDebugEnabled())
+              logger_.debug("Done streaming " + file);
+        }            
+        catch ( SocketException se )
+        {                        
+            logger_.info(LogUtil.throwableToString(se));
+        }
+        catch ( IOException e )
+        {
+            logConnectAndIOException(e, connection);
+        }
+        catch (Throwable th)
+        {
+            logger_.warn(LogUtil.throwableToString(th));
+        }        
+    }
+    
+    private void logConnectAndIOException(IOException ex, TcpConnection connection)
+    {                    
+        if ( connection != null )
+        {
+            connection.errorClose();
+        }
+        logger_.info(LogUtil.throwableToString(ex));
+    }
+}

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/Header.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/Header.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/Header.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/Header.java Thu Jul 30 15:30:21 2009
@@ -1,182 +1,182 @@
-/**
- * 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.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.util.*;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.cassandra.io.ICompactSerializer;
-import org.apache.cassandra.utils.GuidGenerator;
-
-
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public class Header implements java.io.Serializable
-{
-    static final long serialVersionUID = -3194851946523170022L;
-    private static ICompactSerializer<Header> serializer_;
-    private static AtomicInteger idGen_ = new AtomicInteger(0);
-    
-    static
-    {
-        serializer_ = new HeaderSerializer();        
-    }
-    
-    static ICompactSerializer<Header> serializer()
-    {
-        return serializer_;
-    }
-
-    private EndPoint from_;
-    private String type_;
-    private String verb_;
-    private String messageId_;
-    protected Map<String, byte[]> details_ = new Hashtable<String, byte[]>();
-    
-    Header(String id, EndPoint from, String messageType, String verb)
-    {
-        messageId_ = id;
-        from_ = from;
-        type_ = messageType;
-        verb_ = verb;        
-    }
-    
-    Header(String id, EndPoint from, String messageType, String verb, Map<String, byte[]> details)
-    {
-        this(id, from, messageType, verb);
-        details_ = details;
-    }
-
-    Header(EndPoint from, String messageType, String verb)
-    {
-        messageId_ = Integer.toString(idGen_.incrementAndGet());
-        from_ = from;
-        type_ = messageType;
-        verb_ = verb;
-    }        
-
-    EndPoint getFrom()
-    {
-        return from_;
-    }
-
-    String getMessageType()
-    {
-        return type_;
-    }
-
-    String getVerb()
-    {
-        return verb_;
-    }
-
-    String getMessageId()
-    {
-        return messageId_;
-    }
-
-    void setMessageId(String id)
-    {
-        messageId_ = id;
-    }
-    
-    void setMessageType(String type)
-    {
-        type_ = type;
-    }
-    
-    void setMessageVerb(String verb)
-    {
-        verb_ = verb;
-    }
-    
-    byte[] getDetail(Object key)
-    {
-        return details_.get(key);
-    }
-    
-    void removeDetail(Object key)
-    {
-        details_.remove(key);
-    }
-    
-    void addDetail(String key, byte[] value)
-    {
-        details_.put(key, value);
-    }
-    
-    Map<String, byte[]> getDetails()
-    {
-        return details_;
-    }
-}
-
-class HeaderSerializer implements ICompactSerializer<Header>
-{
-    public void serialize(Header t, DataOutputStream dos) throws IOException
-    {           
-        dos.writeUTF(t.getMessageId());
-        CompactEndPointSerializationHelper.serialize(t.getFrom(), dos);
-        dos.writeUTF(t.getMessageType());
-        dos.writeUTF( t.getVerb() );
-        
-        /* Serialize the message header */
-        int size = t.details_.size();
-        dos.writeInt(size);
-        Set<String> keys = t.details_.keySet();
-        
-        for( String key : keys )
-        {
-            dos.writeUTF(key);
-            byte[] value = t.details_.get(key);
-            dos.writeInt(value.length);
-            dos.write(value);
-        }
-    }
-
-    public Header deserialize(DataInputStream dis) throws IOException
-    {
-        String id = dis.readUTF();
-        EndPoint from = CompactEndPointSerializationHelper.deserialize(dis);
-        String type = dis.readUTF();
-        String verb = dis.readUTF();
-        
-        /* Deserializing the message header */
-        int size = dis.readInt();
-        Map<String, byte[]> details = new Hashtable<String, byte[]>(size);
-        for ( int i = 0; i < size; ++i )
-        {
-            String key = dis.readUTF();
-            int length = dis.readInt();
-            byte[] bytes = new byte[length];
-            dis.readFully(bytes);
-            details.put(key, bytes);
-        }
-        
-        return new Header(id, from, type, verb, details);
-    }
-}
-
-
+/**
+ * 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.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.utils.GuidGenerator;
+
+
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class Header implements java.io.Serializable
+{
+    static final long serialVersionUID = -3194851946523170022L;
+    private static ICompactSerializer<Header> serializer_;
+    private static AtomicInteger idGen_ = new AtomicInteger(0);
+    
+    static
+    {
+        serializer_ = new HeaderSerializer();        
+    }
+    
+    static ICompactSerializer<Header> serializer()
+    {
+        return serializer_;
+    }
+
+    private EndPoint from_;
+    private String type_;
+    private String verb_;
+    private String messageId_;
+    protected Map<String, byte[]> details_ = new Hashtable<String, byte[]>();
+    
+    Header(String id, EndPoint from, String messageType, String verb)
+    {
+        messageId_ = id;
+        from_ = from;
+        type_ = messageType;
+        verb_ = verb;        
+    }
+    
+    Header(String id, EndPoint from, String messageType, String verb, Map<String, byte[]> details)
+    {
+        this(id, from, messageType, verb);
+        details_ = details;
+    }
+
+    Header(EndPoint from, String messageType, String verb)
+    {
+        messageId_ = Integer.toString(idGen_.incrementAndGet());
+        from_ = from;
+        type_ = messageType;
+        verb_ = verb;
+    }        
+
+    EndPoint getFrom()
+    {
+        return from_;
+    }
+
+    String getMessageType()
+    {
+        return type_;
+    }
+
+    String getVerb()
+    {
+        return verb_;
+    }
+
+    String getMessageId()
+    {
+        return messageId_;
+    }
+
+    void setMessageId(String id)
+    {
+        messageId_ = id;
+    }
+    
+    void setMessageType(String type)
+    {
+        type_ = type;
+    }
+    
+    void setMessageVerb(String verb)
+    {
+        verb_ = verb;
+    }
+    
+    byte[] getDetail(Object key)
+    {
+        return details_.get(key);
+    }
+    
+    void removeDetail(Object key)
+    {
+        details_.remove(key);
+    }
+    
+    void addDetail(String key, byte[] value)
+    {
+        details_.put(key, value);
+    }
+    
+    Map<String, byte[]> getDetails()
+    {
+        return details_;
+    }
+}
+
+class HeaderSerializer implements ICompactSerializer<Header>
+{
+    public void serialize(Header t, DataOutputStream dos) throws IOException
+    {           
+        dos.writeUTF(t.getMessageId());
+        CompactEndPointSerializationHelper.serialize(t.getFrom(), dos);
+        dos.writeUTF(t.getMessageType());
+        dos.writeUTF( t.getVerb() );
+        
+        /* Serialize the message header */
+        int size = t.details_.size();
+        dos.writeInt(size);
+        Set<String> keys = t.details_.keySet();
+        
+        for( String key : keys )
+        {
+            dos.writeUTF(key);
+            byte[] value = t.details_.get(key);
+            dos.writeInt(value.length);
+            dos.write(value);
+        }
+    }
+
+    public Header deserialize(DataInputStream dis) throws IOException
+    {
+        String id = dis.readUTF();
+        EndPoint from = CompactEndPointSerializationHelper.deserialize(dis);
+        String type = dis.readUTF();
+        String verb = dis.readUTF();
+        
+        /* Deserializing the message header */
+        int size = dis.readInt();
+        Map<String, byte[]> details = new Hashtable<String, byte[]>(size);
+        for ( int i = 0; i < size; ++i )
+        {
+            String key = dis.readUTF();
+            int length = dis.readInt();
+            byte[] bytes = new byte[length];
+            dis.readFully(bytes);
+            details.put(key, bytes);
+        }
+        
+        return new Header(id, from, type, verb, details);
+    }
+}
+
+

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/HeaderTypes.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/HeaderTypes.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/HeaderTypes.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/HeaderTypes.java Thu Jul 30 15:30:21 2009
@@ -1,28 +1,28 @@
-/**
- * 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;
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public class HeaderTypes 
-{
-    public final static String TASK_PROFILE_CHAIN = "TASK_PROFILE_CHAIN";
-    public static String TASK_ID = "TASK_ID";
-}
+/**
+ * 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;
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class HeaderTypes 
+{
+    public final static String TASK_PROFILE_CHAIN = "TASK_PROFILE_CHAIN";
+    public static String TASK_ID = "TASK_ID";
+}

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncCallback.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncCallback.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncCallback.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncCallback.java Thu Jul 30 15:30:21 2009
@@ -1,38 +1,38 @@
-/**
- * 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;
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IAsyncCallback 
-{
-	/**
-	 * @param msg responses to be returned
-	 */
-	public void response(Message msg);
-    
-    /**
-     * Attach some application specific context to the
-     * callback.
-     * @param o application specific context
-     */
-    public void attachContext(Object o);
-}
+/**
+ * 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;
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IAsyncCallback 
+{
+	/**
+	 * @param msg responses to be returned
+	 */
+	public void response(Message msg);
+    
+    /**
+     * Attach some application specific context to the
+     * callback.
+     * @param o application specific context
+     */
+    public void attachContext(Object o);
+}

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncResult.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncResult.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncResult.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IAsyncResult.java Thu Jul 30 15:30:21 2009
@@ -1,73 +1,73 @@
-/**
- * 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.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-/**
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IAsyncResult
-{    
-    /**
-     * This is used to check if the task has been completed
-     * 
-     * @return true if the task has been completed and false otherwise.
-     */
-    public boolean isDone();
-    
-    /**
-     * Returns the result for the task that was submitted.
-     * @return the result wrapped in an Object[]
-    */
-    public byte[] get();
-    
-    /**
-     * 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 byte[] get(long timeout, TimeUnit tu) throws TimeoutException;
-    
-    /**
-     * Returns the result for all tasks that was submitted.
-     * @return the list of results wrapped in an Object[]
-    */
-    public List<byte[]> multiget();
-    
-    /**
-     * 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 List<byte[]> multiget(long timeout, TimeUnit tu) throws TimeoutException;
-    
-    /**
-     * Store the result obtained for the submitted task.
-     * @param result the response message
-     */
-    public void result(Message result);
-}
+/**
+ * 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.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IAsyncResult
+{    
+    /**
+     * This is used to check if the task has been completed
+     * 
+     * @return true if the task has been completed and false otherwise.
+     */
+    public boolean isDone();
+    
+    /**
+     * Returns the result for the task that was submitted.
+     * @return the result wrapped in an Object[]
+    */
+    public byte[] get();
+    
+    /**
+     * 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 byte[] get(long timeout, TimeUnit tu) throws TimeoutException;
+    
+    /**
+     * Returns the result for all tasks that was submitted.
+     * @return the list of results wrapped in an Object[]
+    */
+    public List<byte[]> multiget();
+    
+    /**
+     * 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 List<byte[]> multiget(long timeout, TimeUnit tu) throws TimeoutException;
+    
+    /**
+     * Store the result obtained for the submitted task.
+     * @param result the response message
+     */
+    public void result(Message result);
+}

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IMessagingService.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IMessagingService.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IMessagingService.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IMessagingService.java Thu Jul 30 15:30:21 2009
@@ -1,179 +1,179 @@
-/**
- * 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.io.IOException;
-
-import javax.xml.bind.JAXBException;
-
-import org.apache.cassandra.concurrent.IStage;
-
-
-/**
- * An IMessagingService provides the methods for sending messages to remote
- * endpoints. IMessagingService enables the sending of request-response style
- * messages and fire-forget style messages.
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IMessagingService
-{   
-	/**
-     * Register a verb and the corresponding verb handler with the
-     * Messaging Service.
-     * @param type name of the verb.     
-     * @param verbHandler handler for the specified verb
-     */
-    public void registerVerbHandlers(String type, IVerbHandler verbHandler);
-    
-    /**
-     * Deregister all verbhandlers corresponding to localEndPoint.
-     * @param localEndPoint
-     */
-    public void deregisterAllVerbHandlers(EndPoint localEndPoint);
-    
-    /**
-     * Deregister a verbhandler corresponding to the verb from the
-     * Messaging Service.
-     * @param type name of the verb.      
-     */
-    public void deregisterVerbHandlers(String type);
-    
-    /**
-     * Listen on the specified port.
-     * @param ep EndPoint whose port to listen on.
-     * @param isHttp specify if the port is an Http port.     
-     */
-    public void listen(EndPoint ep) throws IOException;
-    
-    /**
-     * Listen on the specified port.
-     * @param ep EndPoint whose port to listen on.     
-     */
-    public void listenUDP(EndPoint ep);
-    
-    /**
-     * Send a message to a given endpoint. 
-     * @param message message to be sent.
-     * @param to endpoint to which the message needs to be sent
-     * @return an reference to an IAsyncResult which can be queried for the
-     * response
-     */
-    public IAsyncResult sendRR(Message message, EndPoint to);
-
-    /**
-     * Send a message to the given set of endpoints and informs the MessagingService
-     * to wait for at least <code>howManyResults</code> responses to determine success
-     * of failure.
-     * @param message message to be sent.
-     * @param to endpoints to which the message needs to be sent
-     * @param cb callback interface which is used to pass the responses
-     * @return an reference to message id used to match with the result
-     */
-    public String sendRR(Message message, EndPoint[] to, IAsyncCallback cb);
-    
-    /**
-     * Send a message to a given endpoint. This method specifies a callback
-     * which is invoked with the actual response.
-     * @param message message to be sent.
-     * @param to endpoint to which the message needs to be sent
-     * @param cb callback interface which is used to pass the responses or
-     *           suggest that a timeout occurred to the invoker of the send().
-     *           suggest that a timeout occurred to the invoker of the send().
-     * @return an reference to message id used to match with the result
-     */
-    public String sendRR(Message message, EndPoint to, IAsyncCallback cb);
-
-    /**
-     * Send a message to a given endpoint. The ith element in the <code>messages</code>
-     * array is sent to the ith element in the <code>to</code> array.This method assumes
-     * there is a one-one mapping between the <code>messages</code> array and
-     * the <code>to</code> array. Otherwise an  IllegalArgumentException will be thrown.
-     * This method also informs the MessagingService to wait for at least
-     * <code>howManyResults</code> responses to determine success of failure.
-     * @param messages messages to be sent.
-     * @param to endpoints to which the message needs to be sent
-     * @param cb callback interface which is used to pass the responses or
-     *           suggest that a timeout occured to the invoker of the send().
-     * @return an reference to message id used to match with the result
-     */
-    public String sendRR(Message[] messages, EndPoint[] to, IAsyncCallback cb);
-    
-    /**
-     * Send a message to a given endpoint. The ith element in the <code>messages</code>
-     * array is sent to the ith element in the <code>to</code> array.This method assumes
-     * there is a one-one mapping between the <code>messages</code> array and
-     * the <code>to</code> array. Otherwise an  IllegalArgumentException will be thrown.
-     * This method also informs the MessagingService to wait for at least
-     * <code>howManyResults</code> responses to determine success of failure.
-     * @param messages messages to be sent.
-     * @param to endpoints to which the message needs to be sent
-     * @return an reference to IAsyncResult
-     */
-    public IAsyncResult sendRR(Message[] messages, EndPoint[] to);
-    
-    /**
-     * Send a message to a given endpoint. The ith element in the <code>messages</code>
-     * array is sent to the ith element in the <code>to</code> array.This method assumes
-     * there is a one-one mapping between the <code>messages</code> array and
-     * the <code>to</code> array. Otherwise an  IllegalArgumentException will be thrown.
-     * The idea is that multi-groups of messages are grouped as one logical message
-     * whose results are harnessed via the <i>IAsyncResult</i>
-     * @param messages groups of grouped messages.
-     * @param to destination for the groups of messages
-     * @param cb the callback handler to be invoked for the responses
-     * @return the group id which is basically useless - it is only returned for API's
-     *         to look compatible.
-     */
-    public String sendRR(Message[][] messages, EndPoint[][] to, IAsyncCallback cb);
-
-    /**
-     * Send a message to a given endpoint. This method adheres to the fire and forget
-     * style messaging.
-     * @param message messages to be sent.
-     * @param to endpoint to which the message needs to be sent
-     */
-    public void sendOneWay(Message message, EndPoint to);
-        
-    /**
-     * Send a message to a given endpoint. This method adheres to the fire and forget
-     * style messaging.
-     * @param message messages to be sent.
-     * @param to endpoint to which the message needs to be sent
-     */
-    public void sendUdpOneWay(Message message, EndPoint to);
-    
-    /**
-     * Stream a file from source to destination. This is highly optimized
-     * to not hold any of the contents of the file in memory.
-     * @param file name of file to stream.
-     * @param startPosition position inside the file
-     * @param total number of bytes to stream
-     * @param to endpoint to which we need to stream the file.
-    */
-    public void stream(String file, long startPosition, long total, EndPoint from, EndPoint to);
-
-    /**
-     * This method returns the verb handler associated with the registered
-     * verb. If no handler has been registered then null is returned.
-     * @param verb for which the verb handler is sought
-     * @return a reference to IVerbHandler which is the handler for the specified verb
-     */
-    public IVerbHandler getVerbHandler(String verb);    
-}
+/**
+ * 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.io.IOException;
+
+import javax.xml.bind.JAXBException;
+
+import org.apache.cassandra.concurrent.IStage;
+
+
+/**
+ * An IMessagingService provides the methods for sending messages to remote
+ * endpoints. IMessagingService enables the sending of request-response style
+ * messages and fire-forget style messages.
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IMessagingService
+{   
+	/**
+     * Register a verb and the corresponding verb handler with the
+     * Messaging Service.
+     * @param type name of the verb.     
+     * @param verbHandler handler for the specified verb
+     */
+    public void registerVerbHandlers(String type, IVerbHandler verbHandler);
+    
+    /**
+     * Deregister all verbhandlers corresponding to localEndPoint.
+     * @param localEndPoint
+     */
+    public void deregisterAllVerbHandlers(EndPoint localEndPoint);
+    
+    /**
+     * Deregister a verbhandler corresponding to the verb from the
+     * Messaging Service.
+     * @param type name of the verb.      
+     */
+    public void deregisterVerbHandlers(String type);
+    
+    /**
+     * Listen on the specified port.
+     * @param ep EndPoint whose port to listen on.
+     * @param isHttp specify if the port is an Http port.     
+     */
+    public void listen(EndPoint ep) throws IOException;
+    
+    /**
+     * Listen on the specified port.
+     * @param ep EndPoint whose port to listen on.     
+     */
+    public void listenUDP(EndPoint ep);
+    
+    /**
+     * Send a message to a given endpoint. 
+     * @param message message to be sent.
+     * @param to endpoint to which the message needs to be sent
+     * @return an reference to an IAsyncResult which can be queried for the
+     * response
+     */
+    public IAsyncResult sendRR(Message message, EndPoint to);
+
+    /**
+     * Send a message to the given set of endpoints and informs the MessagingService
+     * to wait for at least <code>howManyResults</code> responses to determine success
+     * of failure.
+     * @param message message to be sent.
+     * @param to endpoints to which the message needs to be sent
+     * @param cb callback interface which is used to pass the responses
+     * @return an reference to message id used to match with the result
+     */
+    public String sendRR(Message message, EndPoint[] to, IAsyncCallback cb);
+    
+    /**
+     * Send a message to a given endpoint. This method specifies a callback
+     * which is invoked with the actual response.
+     * @param message message to be sent.
+     * @param to endpoint to which the message needs to be sent
+     * @param cb callback interface which is used to pass the responses or
+     *           suggest that a timeout occurred to the invoker of the send().
+     *           suggest that a timeout occurred to the invoker of the send().
+     * @return an reference to message id used to match with the result
+     */
+    public String sendRR(Message message, EndPoint to, IAsyncCallback cb);
+
+    /**
+     * Send a message to a given endpoint. The ith element in the <code>messages</code>
+     * array is sent to the ith element in the <code>to</code> array.This method assumes
+     * there is a one-one mapping between the <code>messages</code> array and
+     * the <code>to</code> array. Otherwise an  IllegalArgumentException will be thrown.
+     * This method also informs the MessagingService to wait for at least
+     * <code>howManyResults</code> responses to determine success of failure.
+     * @param messages messages to be sent.
+     * @param to endpoints to which the message needs to be sent
+     * @param cb callback interface which is used to pass the responses or
+     *           suggest that a timeout occured to the invoker of the send().
+     * @return an reference to message id used to match with the result
+     */
+    public String sendRR(Message[] messages, EndPoint[] to, IAsyncCallback cb);
+    
+    /**
+     * Send a message to a given endpoint. The ith element in the <code>messages</code>
+     * array is sent to the ith element in the <code>to</code> array.This method assumes
+     * there is a one-one mapping between the <code>messages</code> array and
+     * the <code>to</code> array. Otherwise an  IllegalArgumentException will be thrown.
+     * This method also informs the MessagingService to wait for at least
+     * <code>howManyResults</code> responses to determine success of failure.
+     * @param messages messages to be sent.
+     * @param to endpoints to which the message needs to be sent
+     * @return an reference to IAsyncResult
+     */
+    public IAsyncResult sendRR(Message[] messages, EndPoint[] to);
+    
+    /**
+     * Send a message to a given endpoint. The ith element in the <code>messages</code>
+     * array is sent to the ith element in the <code>to</code> array.This method assumes
+     * there is a one-one mapping between the <code>messages</code> array and
+     * the <code>to</code> array. Otherwise an  IllegalArgumentException will be thrown.
+     * The idea is that multi-groups of messages are grouped as one logical message
+     * whose results are harnessed via the <i>IAsyncResult</i>
+     * @param messages groups of grouped messages.
+     * @param to destination for the groups of messages
+     * @param cb the callback handler to be invoked for the responses
+     * @return the group id which is basically useless - it is only returned for API's
+     *         to look compatible.
+     */
+    public String sendRR(Message[][] messages, EndPoint[][] to, IAsyncCallback cb);
+
+    /**
+     * Send a message to a given endpoint. This method adheres to the fire and forget
+     * style messaging.
+     * @param message messages to be sent.
+     * @param to endpoint to which the message needs to be sent
+     */
+    public void sendOneWay(Message message, EndPoint to);
+        
+    /**
+     * Send a message to a given endpoint. This method adheres to the fire and forget
+     * style messaging.
+     * @param message messages to be sent.
+     * @param to endpoint to which the message needs to be sent
+     */
+    public void sendUdpOneWay(Message message, EndPoint to);
+    
+    /**
+     * Stream a file from source to destination. This is highly optimized
+     * to not hold any of the contents of the file in memory.
+     * @param file name of file to stream.
+     * @param startPosition position inside the file
+     * @param total number of bytes to stream
+     * @param to endpoint to which we need to stream the file.
+    */
+    public void stream(String file, long startPosition, long total, EndPoint from, EndPoint to);
+
+    /**
+     * This method returns the verb handler associated with the registered
+     * verb. If no handler has been registered then null is returned.
+     * @param verb for which the verb handler is sought
+     * @return a reference to IVerbHandler which is the handler for the specified verb
+     */
+    public IVerbHandler getVerbHandler(String verb);    
+}

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IVerbHandler.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IVerbHandler.java?rev=799331&r1=799330&r2=799331&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IVerbHandler.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/net/IVerbHandler.java Thu Jul 30 15:30:21 2009
@@ -1,39 +1,39 @@
-/**
- * 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;
-
-/**
- * IVerbHandler provides the method that all verb handlers need to implement.
- * The concrete implementation of this interface would provide the functionality
- * for a given verb.
- * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
- */
-
-public interface IVerbHandler
-{
-    /**
-     * This method delivers a message to the implementing class (if the implementing
-     * class was registered by a call to MessagingService.registerVerbHandlers).
-     * Note that the caller should not be holding any locks when calling this method
-     * because the implementation may be synchronized.
-     * 
-     * @param message - incoming message that needs handling.     
-     */
-    public void doVerb(Message message);
-}
+/**
+ * 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;
+
+/**
+ * IVerbHandler provides the method that all verb handlers need to implement.
+ * The concrete implementation of this interface would provide the functionality
+ * for a given verb.
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public interface IVerbHandler
+{
+    /**
+     * This method delivers a message to the implementing class (if the implementing
+     * class was registered by a call to MessagingService.registerVerbHandlers).
+     * Note that the caller should not be holding any locks when calling this method
+     * because the implementation may be synchronized.
+     * 
+     * @param message - incoming message that needs handling.     
+     */
+    public void doVerb(Message message);
+}



Mime
View raw message