Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Enable Basic SSL support #6

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Enable Basic SSL support #6

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c85c80e
Remove EOL Whitespace
daurnimator Apr 29, 2013
63c62b5
Uncomment SSL option
daurnimator Apr 29, 2013
d7aff8e
Move SSL config option to Constants
daurnimator Apr 29, 2013
e689606
getFactory shouldn't throw. Use a RuntimeException instead.
daurnimator Apr 30, 2013
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/main/java/org/apache/flume/RabbitMQConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ public class RabbitMQConstants {
public final static String COUNTER_GET_MISS="rabbitmq.getmiss";
public final static String COUNTER_ACK="rabbitmq.ack";
public final static String COUNTER_PUBLISH="rabbitmq.publish";

public final static String CONFIG_HOSTNAME="hostname";
public final static String CONFIG_PORT="port";
public final static String CONFIG_USERNAME="username";
Expand All @@ -36,4 +36,5 @@ public class RabbitMQConstants {
public final static String CONFIG_EXCHANGENAME="exchangename";
public final static String CONFIG_QUEUENAME="queuename";
public final static String CONFIG_TOPICS="topics";
public final static String CONFIG_SSL="usessl";
}
76 changes: 40 additions & 36 deletions src/main/java/org/apache/flume/RabbitMQUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import java.util.Date;
import java.util.Map;

import java.lang.RuntimeException;

import org.apache.commons.collections.map.CaseInsensitiveMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -35,20 +37,20 @@
public class RabbitMQUtil {
private static final Logger log = LoggerFactory.getLogger(RabbitMQUtil.class);
static final String PREFIX="RabbitMQ";

private static void setTimestamp(Map<String,String> headers, BasicProperties properties){
Date date = properties.getTimestamp()==null?new Date():properties.getTimestamp();
Date date = properties.getTimestamp()==null?new Date():properties.getTimestamp();
Long value=date.getTime();
headers.put("timestamp", value.toString());
}

public static Map<String,String> getHeaders(BasicProperties properties){
Preconditions.checkArgument(properties!=null, "properties cannot be null.");
Map<String,String> headers = new CaseInsensitiveMap();
setTimestamp(headers, properties);

Map<String, Object> rabbitmqHeaders = properties.getHeaders();

if(null!=rabbitmqHeaders){
for(Map.Entry<String, Object> kvp:rabbitmqHeaders.entrySet()){
if(!headers.containsKey(kvp.getKey())&&null!=kvp.getValue()){
Expand All @@ -57,82 +59,84 @@ public static Map<String,String> getHeaders(BasicProperties properties){
}
}
}

return headers;
}

public static String getQueueName(Context context) {
return context.getString(RabbitMQConstants.CONFIG_QUEUENAME, "");
//Preconditions.checkState(queueName!=null, "No queueName specified.");
}

public static String getExchangeName(Context context){
return context.getString(RabbitMQConstants.CONFIG_EXCHANGENAME, "");
}

public static String[] getTopics( Context context ) {
String list = context.getString( RabbitMQConstants.CONFIG_TOPICS, "" );

if ( !list.equals("") ) {
return list.split(",");
}

return null;
}
public static ConnectionFactory getFactory(Context context){

public static ConnectionFactory getFactory(Context context) {
Preconditions.checkArgument(context!=null, "context cannot be null.");
ConnectionFactory factory = new ConnectionFactory();

String hostname = context.getString("hostname");
Preconditions.checkArgument(hostname!=null, "No hostname specified.");
factory.setHost(hostname);

int port = context.getInteger(RabbitMQConstants.CONFIG_PORT, -1);

if(-1!=port){
factory.setPort(port);
}

String username = context.getString(RabbitMQConstants.CONFIG_USERNAME);

if(null==username){
factory.setUsername(ConnectionFactory.DEFAULT_USER);
} else {
factory.setUsername(username);
}

String password = context.getString(RabbitMQConstants.CONFIG_PASSWORD);

if(null==password){
factory.setPassword(ConnectionFactory.DEFAULT_PASS);
} else {
factory.setPassword(password);
}

String virtualHost = context.getString(RabbitMQConstants.CONFIG_VIRTUALHOST);

if(null!=virtualHost){
factory.setVirtualHost(virtualHost);
}

int connectionTimeout = context.getInteger(RabbitMQConstants.CONFIG_CONNECTIONTIMEOUT, -1);

if(connectionTimeout>-1){
factory.setConnectionTimeout(connectionTimeout);
factory.setConnectionTimeout(connectionTimeout);
}



// boolean useSSL = context.getBoolean("usessl", false);
// if(useSSL){
// factory.useSslProtocol();
// }



boolean useSSL = context.getBoolean(RabbitMQConstants.CONFIG_SSL, false);

if(useSSL){
try {
factory.useSslProtocol();
} catch (Exception e) {
throw new RuntimeException(e);
}
}

return factory;
}

public static void close(Connection connection, com.rabbitmq.client.Channel channel){
if(null!=channel) {
try {
Expand All @@ -141,7 +145,7 @@ public static void close(Connection connection, com.rabbitmq.client.Channel chan
if(log.isErrorEnabled())log.error("Exception thrown while closing channel", ex);
}
}

if(null!=connection) {
try {
connection.close();
Expand Down