natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

r/m redundant interface IMessagingService. patch by jbellis 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@880926 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-11-16 19:37:12 +00:00
parent b7a1c44333
commit d5cdd6d043
2 changed files with 84 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

151
src/java/org/apache/cassandra/net/IMessagingService.java
View File

@ -1,151 +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.io.IOException;
import java.net.InetAddress;
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.
*/
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(InetAddress 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 InetAddress whose port to listen on.
* @param isHttp specify if the port is an Http port.
*/
public void listen(InetAddress ep) throws IOException;
/**
* Listen on the specified port.
* @param ep InetAddress whose port to listen on.
*/
public void listenUDP(InetAddress 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, InetAddress 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, InetAddress[] 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, InetAddress 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, InetAddress[] 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, InetAddress 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, InetAddress 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, InetAddress from, InetAddress 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);
}

93
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -41,7 +41,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
public class MessagingService implements IMessagingService
public class MessagingService
{
private static int version_ = 1;
//TODO: make this parameter dynamic somehow. Not sure if config is appropriate.
@ -89,7 +89,7 @@ public class MessagingService implements IMessagingService
private static Logger logger_ = Logger.getLogger(MessagingService.class);
private static IMessagingService messagingService_ = new MessagingService();
private static MessagingService messagingService_ = new MessagingService();
private static final int MESSAGE_DESERIALIZE_THREADS = 4;
@ -98,7 +98,7 @@ public class MessagingService implements IMessagingService
return version_;
}
public static IMessagingService instance()
public static MessagingService instance()
{
if ( bShutdown_ )
{
@ -184,6 +184,10 @@ public class MessagingService implements IMessagingService
return result;
}
/**
* Listen on the specified port.
* @param localEp InetAddress whose port to listen on.
*/
public void listen(InetAddress localEp) throws IOException
{
ServerSocketChannel serverChannel = ServerSocketChannel.open();
@ -198,6 +202,10 @@ public class MessagingService implements IMessagingService
listenSockets_.put(localEp, key);
}
/**
* Listen on the specified port.
* @param localEp InetAddress whose port to listen on.
*/
public void listenUDP(InetAddress localEp)
{
UdpConnection connection = new UdpConnection();
@ -251,12 +259,22 @@ public class MessagingService implements 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)
{
checkForReservedVerb(type);
verbHandlers_.put(type, verbHandler);
}
/**
* Deregister all verbhandlers corresponding to localEndPoint.
* @param localEndPoint
*/
public void deregisterAllVerbHandlers(InetAddress localEndPoint)
{
Iterator keys = verbHandlers_.keySet().iterator();
@ -274,17 +292,34 @@ public class MessagingService implements IMessagingService
}
}
/**
* Deregister a verbhandler corresponding to the verb from the
* Messaging Service.
* @param type name of the verb.
*/
public void deregisterVerbHandlers(String type)
{
verbHandlers_.remove(type);
}
/**
* This method returns the verb handler associated with the registered
* verb. If no handler has been registered then null is returned.
* @param type for which the verb handler is sought
* @return a reference to IVerbHandler which is the handler for the specified verb
*/
public IVerbHandler getVerbHandler(String type)
{
IVerbHandler handler = (IVerbHandler)verbHandlers_.get(type);
return handler;
return verbHandlers_.get(type);
}
/**
* 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 String sendRR(Message message, InetAddress[] to, IAsyncCallback cb)
{
String messageId = message.getMessageId();
@ -296,6 +331,16 @@ public class MessagingService implements IMessagingService
return messageId;
}
/**
* 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, InetAddress to, IAsyncCallback cb)
{
String messageId = message.getMessageId();
@ -304,6 +349,19 @@ public class MessagingService implements IMessagingService
return messageId;
}
/**
* 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, InetAddress[] to, IAsyncCallback cb)
{
if ( messages.length != to.length )
@ -320,9 +378,12 @@ public class MessagingService implements IMessagingService
return groupId;
}
/*
Use this version for fire and forget style messaging.
*/
/**
* 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, InetAddress to)
{
// do local deliveries
@ -374,6 +435,12 @@ public class MessagingService implements IMessagingService
return iar;
}
/**
* 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, InetAddress to)
{
if (message.getFrom().equals(to)) {
@ -398,7 +465,15 @@ public class MessagingService implements IMessagingService
connection.close();
}
}
/**
* 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, InetAddress from, InetAddress to)
{
isStreaming_.set(true);