hadoop-zookeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From maha...@apache.org
Subject svn commit: r987314 [5/16] - in /hadoop/zookeeper/trunk: ./ src/contrib/hedwig/ src/contrib/hedwig/client/ src/contrib/hedwig/client/src/ src/contrib/hedwig/client/src/main/ src/contrib/hedwig/client/src/main/cpp/ src/contrib/hedwig/client/src/main/cpp...
Date Thu, 19 Aug 2010 21:25:22 GMT
Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/benchmark/HedwigBenchmark.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,127 @@
+/**
+ * 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.hedwig.client.benchmark;
+
+import java.io.File;
+import java.util.concurrent.Callable;
+
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.log4j.Logger;
+import org.jboss.netty.logging.InternalLoggerFactory;
+import org.jboss.netty.logging.Log4JLoggerFactory;
+
+import com.google.protobuf.ByteString;
+import org.apache.hedwig.client.conf.ClientConfiguration;
+import org.apache.hedwig.client.netty.HedwigClient;
+import org.apache.hedwig.client.netty.HedwigPublisher;
+import org.apache.hedwig.client.netty.HedwigSubscriber;
+
+public class HedwigBenchmark implements Callable<Void> {
+    protected static final Logger logger = Logger.getLogger(HedwigBenchmark.class);
+
+    static final String TOPIC_PREFIX = "topic-";
+
+    private final HedwigClient client;
+    private final HedwigPublisher publisher;
+    private final HedwigSubscriber subscriber;
+
+    public HedwigBenchmark(ClientConfiguration cfg) {
+        client = new HedwigClient(cfg);
+        publisher = client.getPublisher();
+        subscriber = client.getSubscriber();
+    }
+
+    static boolean amIResponsibleForTopic(int topicNum, int partitionIndex, int numPartitions) {
+        return topicNum % numPartitions == partitionIndex;
+    }
+
+    @Override
+    public Void call() throws Exception {
+
+        //
+        // Parameters.
+        //
+
+        // What program to run: pub, sub (subscription benchmark), recv.
+        final String mode = System.getProperty("mode","");
+
+        // Number of requests to make (publishes or subscribes).
+        int numTopics = Integer.getInteger("nTopics", 50);
+        int numMessages = Integer.getInteger("nMsgs", 1000);
+        int numRegions = Integer.getInteger("nRegions", 1);
+        int startTopicLabel = Integer.getInteger("startTopicLabel", 0);
+        int partitionIndex = Integer.getInteger("partitionIndex", 0);
+        int numPartitions = Integer.getInteger("nPartitions", 1);
+
+        int replicaIndex = Integer.getInteger("replicaIndex", 0);
+
+        int rate = Integer.getInteger("rate", 0);
+        int nParallel = Integer.getInteger("npar", 100);
+        int msgSize = Integer.getInteger("msgSize", 1024);
+
+        // Number of warmup subscriptions to make.
+        final int nWarmups = Integer.getInteger("nwarmups", 1000);
+
+        if (mode.equals("sub")) {
+            BenchmarkSubscriber benchmarkSub = new BenchmarkSubscriber(numTopics, 0, 1, startTopicLabel, 0, 1,
+                    subscriber, ByteString.copyFromUtf8("mySub"));
+
+            benchmarkSub.warmup(nWarmups);
+            benchmarkSub.call();
+
+        } else if (mode.equals("recv")) {
+
+            BenchmarkSubscriber benchmarkSub = new BenchmarkSubscriber(numTopics, numMessages, numRegions,
+                    startTopicLabel, partitionIndex, numPartitions, subscriber, ByteString.copyFromUtf8("sub-"
+                            + replicaIndex));
+
+            benchmarkSub.call();
+
+        } else if (mode.equals("pub")) {
+            // Offered load in msgs/second.
+            BenchmarkPublisher benchmarkPub = new BenchmarkPublisher(numTopics, numMessages, numRegions,
+                    startTopicLabel, partitionIndex, numPartitions, publisher, subscriber, msgSize, nParallel, rate);
+            benchmarkPub.warmup(nWarmups);
+            benchmarkPub.call();
+            
+        } else {
+            throw new Exception("unknown mode: " + mode);
+        }
+
+        return null;
+    }
+
+    public static void main(String[] args) throws Exception {
+        ClientConfiguration cfg = new ClientConfiguration();
+        if (args.length > 0) {
+            String confFile = args[0];
+            try {
+                cfg.loadConf(new File(confFile).toURI().toURL());
+            } catch (ConfigurationException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        InternalLoggerFactory.setDefaultFactory(new Log4JLoggerFactory());
+
+        HedwigBenchmark app = new HedwigBenchmark(cfg);
+        app.call();
+        System.exit(0);
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/conf/ClientConfiguration.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,148 @@
+/**
+ * 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.hedwig.client.conf;
+
+import java.net.InetSocketAddress;
+
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.log4j.Logger;
+
+import org.apache.hedwig.conf.AbstractConfiguration;
+import org.apache.hedwig.util.HedwigSocketAddress;
+
+public class ClientConfiguration extends AbstractConfiguration {
+    Logger logger = Logger.getLogger(ClientConfiguration.class);
+
+    // Protected member variables for configuration parameter names
+    protected static final String DEFAULT_SERVER_HOST = "default_server_host";
+    protected static final String MAX_MESSAGE_SIZE = "max_message_size";
+    protected static final String MAX_SERVER_REDIRECTS = "max_server_redirects";
+    protected static final String AUTO_SEND_CONSUME_MESSAGE_ENABLED = "auto_send_consume_message_enabled";
+    protected static final String CONSUMED_MESSAGES_BUFFER_SIZE = "consumed_messages_buffer_size";
+    protected static final String MESSAGE_CONSUME_RETRY_WAIT_TIME = "message_consume_retry_wait_time";
+    protected static final String SUBSCRIBE_RECONNECT_RETRY_WAIT_TIME = "subscribe_reconnect_retry_wait_time";
+    protected static final String MAX_OUTSTANDING_MESSAGES = "max_outstanding_messages";
+    protected static final String SERVER_ACK_RESPONSE_TIMEOUT = "server_ack_response_timeout";
+    protected static final String TIMEOUT_THREAD_RUN_INTERVAL = "timeout_thread_run_interval";
+    protected static final String SSL_ENABLED = "ssl_enabled";
+
+    // Singletons we want to instantiate only once per ClientConfiguration
+    protected HedwigSocketAddress myDefaultServerAddress = null;
+
+    // Getters for the various Client Configuration parameters.
+    // This should point to the default server host, or the VIP fronting all of
+    // the server hubs. This will return the HedwigSocketAddress which
+    // encapsulates both the regular and SSL port connection to the server host.
+    protected HedwigSocketAddress getDefaultServerHedwigSocketAddress() {
+        if (myDefaultServerAddress == null)
+            myDefaultServerAddress = new HedwigSocketAddress(conf.getString(DEFAULT_SERVER_HOST, "localhost:4080:9876"));
+        return myDefaultServerAddress;
+    }
+
+    // This will get the default server InetSocketAddress based on if SSL is
+    // enabled or not.
+    public InetSocketAddress getDefaultServerHost() {
+        if (isSSLEnabled())
+            return getDefaultServerHedwigSocketAddress().getSSLSocketAddress();
+        else
+            return getDefaultServerHedwigSocketAddress().getSocketAddress();
+    }
+
+    public int getMaximumMessageSize() {
+        return conf.getInt(MAX_MESSAGE_SIZE, 2 * 1024 * 1024);
+    }
+
+    // This parameter is for setting the maximum number of server redirects to
+    // allow before we consider it as an error condition. This is to stop
+    // infinite redirect loops in case there is a problem with the hub servers
+    // topic mastership.
+    public int getMaximumServerRedirects() {
+        return conf.getInt(MAX_SERVER_REDIRECTS, 2);
+    }
+
+    // This parameter is a boolean flag indicating if the client library should
+    // automatically send the consume message to the server based on the
+    // configured amount of messages consumed by the client app. The client app
+    // could choose to override this behavior and instead, manually send the
+    // consume message to the server via the client library using its own 
+    // logic and policy.
+    public boolean isAutoSendConsumeMessageEnabled() {
+        return conf.getBoolean(AUTO_SEND_CONSUME_MESSAGE_ENABLED, true);
+    }
+
+    // This parameter is to set how many consumed messages we'll buffer up
+    // before we send the Consume message to the server indicating that all
+    // of the messages up to that point have been successfully consumed by
+    // the client.
+    public int getConsumedMessagesBufferSize() {
+        return conf.getInt(CONSUMED_MESSAGES_BUFFER_SIZE, 5);
+    }
+
+    // This parameter is used to determine how long we wait before retrying the
+    // client app's MessageHandler to consume a subscribed messages sent to us
+    // from the server. The time to wait is in milliseconds.
+    public long getMessageConsumeRetryWaitTime() {
+        return conf.getLong(MESSAGE_CONSUME_RETRY_WAIT_TIME, 10000);
+    }
+
+    // This parameter is used to determine how long we wait before retrying the
+    // Subscribe Reconnect request. This is done when the connection to a server
+    // disconnects and we attempt to connect to it. We'll keep on trying but
+    // in case the server(s) is down for a longer time, we want to throttle
+    // how often we do the subscribe reconnect request. The time to wait is in
+    // milliseconds.
+    public long getSubscribeReconnectRetryWaitTime() {
+        return conf.getLong(SUBSCRIBE_RECONNECT_RETRY_WAIT_TIME, 10000);
+    }
+
+    // This parameter is for setting the maximum number of outstanding messages
+    // the client app can be consuming at a time for topic subscription before
+    // we throttle things and stop reading from the Netty Channel.
+    public int getMaximumOutstandingMessages() {
+        return conf.getInt(MAX_OUTSTANDING_MESSAGES, 10);
+    }
+
+    // This parameter is used to determine how long we wait (in milliseconds)
+    // before we time out outstanding PubSubRequests that were written to the
+    // server successfully but haven't yet received the ack response.
+    public long getServerAckResponseTimeout() {
+        return conf.getLong(SERVER_ACK_RESPONSE_TIMEOUT, 30000);
+    }
+
+    // This parameter is used to determine how often we run the server ack
+    // response timeout cleaner thread (in milliseconds).
+    public long getTimeoutThreadRunInterval() {
+        return conf.getLong(TIMEOUT_THREAD_RUN_INTERVAL, 60000);
+    }
+
+    // This parameter is a boolean flag indicating if communication with the
+    // server should be done via SSL for encryption. This is needed for
+    // cross-colo hub clients listening to non-local servers.
+    public boolean isSSLEnabled() {
+        return conf.getBoolean(SSL_ENABLED, false);
+    }
+
+    // Validate that the configuration properties are valid.
+    public void validate() throws ConfigurationException {
+        if (isSSLEnabled() && getDefaultServerHedwigSocketAddress().getSSLSocketAddress() == null) {
+            throw new ConfigurationException("SSL is enabled but a default server SSL port not given!");
+        }
+        // Add other validation checks here
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/MessageConsumeData.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,58 @@
+/**
+ * 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.hedwig.client.data;
+
+import com.google.protobuf.ByteString;
+import org.apache.hedwig.protocol.PubSubProtocol.Message;
+
+/**
+ * Wrapper class to store all of the data points needed to encapsulate Message
+ * Consumption in the Subscribe flow for consuming a message sent from the
+ * server for a given TopicSubscriber. This will be used as the Context in the
+ * VoidCallback for the MessageHandlers once they've completed consuming the
+ * message.
+ * 
+ */
+public class MessageConsumeData {
+
+    // Member variables
+    public final ByteString topic;
+    public final ByteString subscriberId;
+    // This is the Message sent from the server for Subscribes for consumption
+    // by the client.
+    public final Message msg;
+
+    // Constructor
+    public MessageConsumeData(final ByteString topic, final ByteString subscriberId, final Message msg) {
+        this.topic = topic;
+        this.subscriberId = subscriberId;
+        this.msg = msg;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+        if (topic != null)
+            sb.append("Topic: " + topic.toStringUtf8());
+        if (subscriberId != null)
+            sb.append(PubSubData.COMMA).append("SubscriberId: " + subscriberId.toStringUtf8());
+        if (msg != null)
+            sb.append(PubSubData.COMMA).append("Message: " + msg);
+        return sb.toString();
+    }
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/PubSubData.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/PubSubData.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/PubSubData.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/PubSubData.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,149 @@
+/**
+ * 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.hedwig.client.data;
+
+import java.util.List;
+
+import com.google.protobuf.ByteString;
+import org.apache.hedwig.protocol.PubSubProtocol.Message;
+import org.apache.hedwig.protocol.PubSubProtocol.OperationType;
+import org.apache.hedwig.protocol.PubSubProtocol.SubscribeRequest.CreateOrAttach;
+import org.apache.hedwig.util.Callback;
+
+/**
+ * Wrapper class to store all of the data points needed to encapsulate all
+ * PubSub type of request operations the client will do. This includes knowing
+ * all of the information needed if we need to redo the publish/subscribe
+ * request in case of a server redirect. This will be used for all sync/async
+ * calls, and for all the known types of request messages to send to the server
+ * hubs: Publish, Subscribe, Unsubscribe, and Consume.
+ * 
+ */
+public class PubSubData {
+    // Static string constants
+    protected static final String COMMA = ", ";
+
+    // Member variables needed during object construction time.
+    public final ByteString topic;
+    public final Message msg;
+    public final ByteString subscriberId;
+    // Enum to indicate what type of operation this PubSub request data object
+    // is for.
+    public final OperationType operationType;
+    // Enum for subscribe requests to indicate if this is a CREATE, ATTACH, or
+    // CREATE_OR_ATTACH subscription request. For non-subscribe requests,
+    // this will be null.
+    public final CreateOrAttach createOrAttach;
+    // These two variables are not final since we might override them
+    // in the case of a Subscribe reconnect.
+    public Callback<Void> callback;
+    public Object context;
+
+    // Member variables used after object has been constructed.
+    // List of all servers we've sent the PubSubRequest to successfully.
+    // This is to keep track of redirected servers that responded back to us.
+    public List<ByteString> triedServers;
+    // List of all servers that we've tried to connect or write to but
+    // was unsuccessful. We'll retry sending the PubSubRequest but will
+    // quit if we're trying to connect or write to a server that we've
+    // attempted to previously.
+    public List<ByteString> connectFailedServers;
+    public List<ByteString> writeFailedServers;
+    // Boolean to the hub server indicating if it should claim ownership
+    // of the topic the PubSubRequest is for. This is mainly used after
+    // a server redirect. Defaults to false.
+    public boolean shouldClaim = false;
+    // TxnID for the PubSubData if it was sent as a PubSubRequest to the hub
+    // server. This is used in the WriteCallback in case of failure. We want
+    // to remove it from the ResponseHandler.txn2PubSubData map since the
+    // failed PubSubRequest will not get an ack response from the server.
+    // This is set later in the PubSub flows only when we write the actual
+    // request. Therefore it is not an argument in the constructor.
+    public long txnId;
+    // Time in milliseconds using the System.currentTimeMillis() call when the
+    // PubSubRequest was written on the netty Channel to the server.
+    public long requestWriteTime;
+    // For synchronous calls, this variable is used to know when the background
+    // async process for it has completed, set in the VoidCallback.
+    public boolean isDone = false;
+
+    // Constructor for all types of PubSub request data to send to the server
+    public PubSubData(final ByteString topic, final Message msg, final ByteString subscriberId,
+            final OperationType operationType, final CreateOrAttach createOrAttach, final Callback<Void> callback,
+            final Object context) {
+        this.topic = topic;
+        this.msg = msg;
+        this.subscriberId = subscriberId;
+        this.operationType = operationType;
+        this.createOrAttach = createOrAttach;
+        this.callback = callback;
+        this.context = context;
+    }
+
+    // Clear all of the stored servers we've contacted or attempted to in this
+    // request.
+    public void clearServersList() {
+        if (triedServers != null)
+            triedServers.clear();
+        if (connectFailedServers != null)
+            connectFailedServers.clear();
+        if (writeFailedServers != null)
+            writeFailedServers.clear();
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+        if (topic != null)
+            sb.append("Topic: " + topic.toStringUtf8());
+        if (msg != null)
+            sb.append(COMMA).append("Message: " + msg);
+        if (subscriberId != null)
+            sb.append(COMMA).append("SubscriberId: " + subscriberId.toStringUtf8());
+        if (operationType != null)
+            sb.append(COMMA).append("Operation Type: " + operationType.toString());
+        if (createOrAttach != null)
+            sb.append(COMMA).append("Create Or Attach: " + createOrAttach.toString());
+        if (triedServers != null && triedServers.size() > 0) {
+            sb.append(COMMA).append("Tried Servers: ");
+            for (ByteString triedServer : triedServers) {
+                sb.append(triedServer.toStringUtf8()).append(COMMA);
+            }
+        }
+        if (connectFailedServers != null && connectFailedServers.size() > 0) {
+            sb.append(COMMA).append("Connect Failed Servers: ");
+            for (ByteString connectFailedServer : connectFailedServers) {
+                sb.append(connectFailedServer.toStringUtf8()).append(COMMA);
+            }
+        }
+        if (writeFailedServers != null && writeFailedServers.size() > 0) {
+            sb.append(COMMA).append("Write Failed Servers: ");
+            for (ByteString writeFailedServer : writeFailedServers) {
+                sb.append(writeFailedServer.toStringUtf8()).append(COMMA);
+            }
+        }
+        sb.append(COMMA).append("Should Claim: " + shouldClaim);
+        if (txnId != 0)
+            sb.append(COMMA).append("TxnID: " + txnId);
+        if (requestWriteTime != 0)
+            sb.append(COMMA).append("Request Write Time: " + requestWriteTime);
+        sb.append(COMMA).append("Is Done: " + isDone);
+        return sb.toString();
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/data/TopicSubscriber.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,74 @@
+/**
+ * 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.hedwig.client.data;
+
+import org.apache.commons.lang.builder.HashCodeBuilder;
+
+import com.google.protobuf.ByteString;
+
+/**
+ * Wrapper class object for the Topic + SubscriberId combination. Since the
+ * Subscribe flows always use the Topic + SubscriberId as the logical entity,
+ * we'll create a simple class to encapsulate that.
+ * 
+ */
+public class TopicSubscriber {
+    private final ByteString topic;
+    private final ByteString subscriberId;
+    private final int hashCode;
+
+    public TopicSubscriber(final ByteString topic, final ByteString subscriberId) {
+        this.topic = topic;
+        this.subscriberId = subscriberId;
+        hashCode = new HashCodeBuilder().append(topic).append(subscriberId).toHashCode();
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (o == this)
+            return true;
+        if (!(o instanceof TopicSubscriber))
+            return false;
+        final TopicSubscriber obj = (TopicSubscriber) o;
+        return topic.equals(obj.topic) && subscriberId.equals(obj.subscriberId);
+    }
+
+    @Override
+    public int hashCode() {
+        return hashCode;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+        if (topic != null)
+            sb.append("Topic: " + topic.toStringUtf8());
+        if (subscriberId != null)
+            sb.append(PubSubData.COMMA).append("SubscriberId: " + subscriberId.toStringUtf8());
+        return sb.toString();
+    }
+    
+    public ByteString getTopic() {
+        return topic;
+    }
+    
+    public ByteString getSubscriberId() {
+        return subscriberId;
+    }
+
+}
\ No newline at end of file

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/InvalidSubscriberIdException.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,37 @@
+/**
+ * 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.hedwig.client.exceptions;
+
+/**
+ * This is a Hedwig client side exception when the local client wants to do
+ * subscribe type of operations. Currently, to distinguish between local and hub
+ * subscribers, the subscriberId will have a specific format.
+ */
+public class InvalidSubscriberIdException extends Exception {
+
+    private static final long serialVersionUID = 873259807218723523L;
+
+    public InvalidSubscriberIdException(String message) {
+        super(message);
+    }
+
+    public InvalidSubscriberIdException(String message, Throwable t) {
+        super(message, t);
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/ServerRedirectLoopException.java Thu Aug 19 21:25:13 2010
@@ -0,0 +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.hedwig.client.exceptions;
+
+/**
+ * This is a Hedwig client side exception when the PubSubRequest is being
+ * redirected to a server where the request has already been sent to previously. 
+ * To avoid having a cyclical redirect loop, this condition is checked for
+ * and this exception will be thrown to the client caller. 
+ */
+public class ServerRedirectLoopException extends Exception {
+
+    private static final long serialVersionUID = 98723508723152897L;
+
+    public ServerRedirectLoopException(String message) {
+        super(message);
+    }
+
+    public ServerRedirectLoopException(String message, Throwable t) {
+        super(message, t);
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/exceptions/TooManyServerRedirectsException.java Thu Aug 19 21:25:13 2010
@@ -0,0 +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.hedwig.client.exceptions;
+
+/**
+ * This is a Hedwig client side exception when there have been too many server
+ * redirects during a publish/subscribe call. We only allow a certain number of
+ * server redirects to find the topic master. If we have exceeded this
+ * configured amount, the publish/subscribe will fail with this exception.
+ * 
+ */
+public class TooManyServerRedirectsException extends Exception {
+
+    private static final long serialVersionUID = 2341192937965635310L;
+
+    public TooManyServerRedirectsException(String message) {
+        super(message);
+    }
+
+    public TooManyServerRedirectsException(String message, Throwable t) {
+        super(message, t);
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/MessageConsumeCallback.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,95 @@
+/**
+ * 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.hedwig.client.handlers;
+
+import java.util.TimerTask;
+
+import org.apache.log4j.Logger;
+import org.jboss.netty.channel.Channel;
+
+import org.apache.hedwig.client.data.MessageConsumeData;
+import org.apache.hedwig.client.data.TopicSubscriber;
+import org.apache.hedwig.client.netty.HedwigClient;
+import org.apache.hedwig.exceptions.PubSubException;
+import org.apache.hedwig.util.Callback;
+
+/**
+ * This is the Callback used by the MessageHandlers on the client app when
+ * they've finished consuming a subscription message sent from the server
+ * asynchronously. This callback back to the client libs will be stateless so we
+ * can use a singleton for the class. The object context used should be the
+ * MessageConsumeData type. That will contain all of the information needed to
+ * call the message consume logic in the client lib ResponseHandler.
+ * 
+ */
+public class MessageConsumeCallback implements Callback<Void> {
+
+    private static Logger logger = Logger.getLogger(MessageConsumeCallback.class);
+
+    private final HedwigClient client;
+
+    public MessageConsumeCallback(HedwigClient client) {
+        this.client = client;
+    }
+
+    class MessageConsumeRetryTask extends TimerTask {
+        private final MessageConsumeData messageConsumeData;
+        private final TopicSubscriber topicSubscriber;
+
+        public MessageConsumeRetryTask(MessageConsumeData messageConsumeData, TopicSubscriber topicSubscriber) {
+            this.messageConsumeData = messageConsumeData;
+            this.topicSubscriber = topicSubscriber;
+        }
+
+        @Override
+        public void run() {
+            // Try to consume the message again
+            Channel topicSubscriberChannel = client.getSubscriber().getChannelForTopic(topicSubscriber);
+            HedwigClient.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
+                    .asyncMessageConsume(messageConsumeData.msg);
+        }
+    }
+
+    public void operationFinished(Object ctx, Void resultOfOperation) {
+        MessageConsumeData messageConsumeData = (MessageConsumeData) ctx;
+        TopicSubscriber topicSubscriber = new TopicSubscriber(messageConsumeData.topic, messageConsumeData.subscriberId);
+        // Message has been successfully consumed by the client app so callback
+        // to the ResponseHandler indicating that the message is consumed.
+        Channel topicSubscriberChannel = client.getSubscriber().getChannelForTopic(topicSubscriber);
+        HedwigClient.getResponseHandlerFromChannel(topicSubscriberChannel).getSubscribeResponseHandler()
+                .messageConsumed(messageConsumeData.msg);
+    }
+
+    public void operationFailed(Object ctx, PubSubException exception) {
+        // Message has NOT been successfully consumed by the client app so
+        // callback to the ResponseHandler to try the async MessageHandler
+        // Consume logic again.
+        MessageConsumeData messageConsumeData = (MessageConsumeData) ctx;
+        TopicSubscriber topicSubscriber = new TopicSubscriber(messageConsumeData.topic, messageConsumeData.subscriberId);
+        logger.error("Message was not consumed successfully by client MessageHandler: " + messageConsumeData);
+
+        // Sleep a pre-configured amount of time (in milliseconds) before we
+        // do the retry. In the future, we can have more dynamic logic on
+        // what duration to sleep based on how many times we've retried, or
+        // perhaps what the last amount of time we slept was. We could stick
+        // some of this meta-data into the MessageConsumeData when we retry.
+        client.getClientTimer().schedule(new MessageConsumeRetryTask(messageConsumeData, topicSubscriber),
+                client.getConfiguration().getMessageConsumeRetryWaitTime());
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PubSubCallback.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,87 @@
+/**
+ * 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.hedwig.client.handlers;
+
+import org.apache.log4j.Logger;
+
+import org.apache.hedwig.client.data.PubSubData;
+import org.apache.hedwig.exceptions.PubSubException;
+import org.apache.hedwig.util.Callback;
+
+/**
+ * This class is used when we are doing synchronous type of operations. All
+ * underlying client ops in Hedwig are async so this is just a way to make the
+ * async calls synchronous.
+ * 
+ */
+public class PubSubCallback implements Callback<Void> {
+
+    private static Logger logger = Logger.getLogger(PubSubCallback.class);
+
+    // Private member variables
+    private PubSubData pubSubData;
+    // Boolean indicator to see if the sync PubSub call was successful or not.
+    private boolean isCallSuccessful;
+    // For sync callbacks, we'd like to know what the PubSubException is thrown
+    // on failure. This is so we can have a handle to the exception and rethrow
+    // it later.
+    private PubSubException failureException;
+
+    // Constructor
+    public PubSubCallback(PubSubData pubSubData) {
+        this.pubSubData = pubSubData;
+    }
+
+    public void operationFinished(Object ctx, Void resultOfOperation) {
+        if (logger.isDebugEnabled())
+            logger.debug("PubSub call succeeded for pubSubData: " + pubSubData);
+        // Wake up the main sync PubSub thread that is waiting for us to
+        // complete.
+        synchronized (pubSubData) {
+            isCallSuccessful = true;
+            pubSubData.isDone = true;
+            pubSubData.notify();
+        }
+    }
+
+    public void operationFailed(Object ctx, PubSubException exception) {
+        if (logger.isDebugEnabled())
+            logger.debug("PubSub call failed with exception: " + exception + ", pubSubData: " + pubSubData);
+        // Wake up the main sync PubSub thread that is waiting for us to
+        // complete.
+        synchronized (pubSubData) {
+            isCallSuccessful = false;
+            failureException = exception;
+            pubSubData.isDone = true;
+            pubSubData.notify();
+        }
+    }
+
+    // Public getter to determine if the PubSub callback is successful or not
+    // based on the PubSub ack response from the server.
+    public boolean getIsCallSuccessful() {
+        return isCallSuccessful;
+    }
+
+    // Public getter to retrieve what the PubSubException was that occurred when
+    // the operation failed.
+    public PubSubException getFailureException() {
+        return failureException;
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/PublishResponseHandler.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,70 @@
+/**
+ * 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.hedwig.client.handlers;
+
+import org.apache.log4j.Logger;
+import org.jboss.netty.channel.Channel;
+
+import org.apache.hedwig.client.data.PubSubData;
+import org.apache.hedwig.client.netty.HedwigClient;
+import org.apache.hedwig.client.netty.ResponseHandler;
+import org.apache.hedwig.exceptions.PubSubException.ServiceDownException;
+import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
+
+public class PublishResponseHandler {
+
+    private static Logger logger = Logger.getLogger(PublishResponseHandler.class);
+
+    private final ResponseHandler responseHandler;
+
+    public PublishResponseHandler(ResponseHandler responseHandler) {
+        this.responseHandler = responseHandler;
+    }
+
+    // Main method to handle Publish Response messages from the server.
+    public void handlePublishResponse(PubSubResponse response, PubSubData pubSubData, Channel channel) throws Exception {
+        if (logger.isDebugEnabled())
+            logger.debug("Handling a Publish response: " + response + ", pubSubData: " + pubSubData + ", host: "
+                    + HedwigClient.getHostFromChannel(channel));
+        switch (response.getStatusCode()) {
+        case SUCCESS:
+            // Response was success so invoke the callback's operationFinished
+            // method.
+            pubSubData.callback.operationFinished(pubSubData.context, null);
+            break;
+        case SERVICE_DOWN:
+            // Response was service down failure so just invoke the callback's
+            // operationFailed method.
+            pubSubData.callback.operationFailed(pubSubData.context, new ServiceDownException(
+                    "Server responded with a SERVICE_DOWN status"));
+            break;
+        case NOT_RESPONSIBLE_FOR_TOPIC:
+            // Redirect response so we'll need to repost the original Publish
+            // Request
+            responseHandler.handleRedirectResponse(response, pubSubData, channel);
+            break;
+        default:
+            // Consider all other status codes as errors, operation failed
+            // cases.
+            logger.error("Unexpected error response from server for PubSubResponse: " + response);
+            pubSubData.callback.operationFailed(pubSubData.context, new ServiceDownException(
+                    "Server responded with a status code of: " + response.getStatusCode()));
+            break;
+        }
+    }
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeReconnectCallback.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeReconnectCallback.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeReconnectCallback.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeReconnectCallback.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,113 @@
+/**
+ * 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.hedwig.client.handlers;
+
+import java.util.TimerTask;
+
+import org.apache.log4j.Logger;
+
+import org.apache.hedwig.client.api.MessageHandler;
+import org.apache.hedwig.client.conf.ClientConfiguration;
+import org.apache.hedwig.client.data.PubSubData;
+import org.apache.hedwig.client.netty.HedwigClient;
+import org.apache.hedwig.client.netty.HedwigSubscriber;
+import org.apache.hedwig.exceptions.PubSubException;
+import org.apache.hedwig.exceptions.PubSubException.ClientNotSubscribedException;
+import org.apache.hedwig.util.Callback;
+
+/**
+ * This class is used when a Subscribe channel gets disconnected and we attempt
+ * to re-establish the connection. Once the connection to the server host for
+ * the topic is completed, we need to restart delivery for that topic if that
+ * was the case before the original channel got disconnected. This async
+ * callback will be the hook for this.
+ * 
+ */
+public class SubscribeReconnectCallback implements Callback<Void> {
+
+    private static Logger logger = Logger.getLogger(SubscribeReconnectCallback.class);
+
+    // Private member variables
+    private final PubSubData origSubData;
+    private final HedwigClient client;
+    private final HedwigSubscriber sub;
+    private final ClientConfiguration cfg;
+    private final MessageHandler messageHandler;
+
+    // Constructor
+    public SubscribeReconnectCallback(PubSubData origSubData, HedwigClient client, MessageHandler messageHandler) {
+        this.origSubData = origSubData;
+        this.client = client;
+        this.sub = client.getSubscriber();
+        this.cfg = client.getConfiguration();
+        this.messageHandler = messageHandler;
+    }
+
+    class SubscribeReconnectRetryTask extends TimerTask {
+        @Override
+        public void run() {
+            if (logger.isDebugEnabled())
+                logger.debug("Retrying subscribe reconnect request for origSubData: " + origSubData);
+            // Clear out all of the servers we've contacted or attempted to from
+            // this request.
+            origSubData.clearServersList();
+            client.doConnect(origSubData, cfg.getDefaultServerHost());
+        }
+    }
+
+    public void operationFinished(Object ctx, Void resultOfOperation) {
+        if (logger.isDebugEnabled())
+            logger.debug("Subscribe reconnect succeeded for origSubData: " + origSubData);
+        // Now we want to restart delivery for the subscription channel only
+        // if delivery was started at the time the original subscribe channel
+        // was disconnected.
+        if (messageHandler != null) {
+            try {
+                sub.startDelivery(origSubData.topic, origSubData.subscriberId, messageHandler);
+            } catch (ClientNotSubscribedException e) {
+                // This exception should never be thrown here but just in case,
+                // log an error and just keep retrying the subscribe request.
+                logger.error("Subscribe was successful but error starting delivery for topic: "
+                        + origSubData.topic.toStringUtf8() + ", subscriberId: "
+                        + origSubData.subscriberId.toStringUtf8(), e);
+                retrySubscribeRequest();
+            }
+        }
+    }
+
+    public void operationFailed(Object ctx, PubSubException exception) {
+        // If the subscribe reconnect fails, just keep retrying the subscribe
+        // request. There isn't a way to flag to the application layer that
+        // a topic subscription has failed. So instead, we'll just keep
+        // retrying in the background until success.
+        logger.error("Subscribe reconnect failed with error: " + exception.getMessage());
+        retrySubscribeRequest();
+    }
+
+    private void retrySubscribeRequest() {
+        // If the client has stopped, there is no need to proceed with any
+        // callback logic here.
+        if (client.hasStopped())
+            return;
+
+        // Retry the subscribe request but only after waiting for a
+        // preconfigured amount of time.
+        client.getClientTimer().schedule(new SubscribeReconnectRetryTask(),
+                client.getConfiguration().getSubscribeReconnectRetryWaitTime());
+    }
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/SubscribeResponseHandler.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,329 @@
+/**
+ * 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.hedwig.client.handlers;
+
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.Set;
+
+import org.apache.log4j.Logger;
+import org.jboss.netty.channel.Channel;
+
+import org.apache.hedwig.client.api.MessageHandler;
+import org.apache.hedwig.client.data.MessageConsumeData;
+import org.apache.hedwig.client.data.PubSubData;
+import org.apache.hedwig.client.data.TopicSubscriber;
+import org.apache.hedwig.client.netty.HedwigClient;
+import org.apache.hedwig.client.netty.ResponseHandler;
+import org.apache.hedwig.exceptions.PubSubException.ClientAlreadySubscribedException;
+import org.apache.hedwig.exceptions.PubSubException.ServiceDownException;
+import org.apache.hedwig.protocol.PubSubProtocol.Message;
+import org.apache.hedwig.protocol.PubSubProtocol.MessageSeqId;
+import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
+import org.apache.hedwig.protocol.PubSubProtocol.StatusCode;
+
+public class SubscribeResponseHandler {
+
+    private static Logger logger = Logger.getLogger(SubscribeResponseHandler.class);
+
+    private final ResponseHandler responseHandler;
+
+    // Member variables used when this ResponseHandler is for a Subscribe
+    // channel. We need to be able to consume messages sent back to us from
+    // the server, and to also recreate the Channel connection if it ever goes
+    // down. For that, we need to store the original PubSubData for the
+    // subscribe request, and also the MessageHandler that was registered when
+    // delivery of messages started for the subscription.
+    private PubSubData origSubData;
+    private Channel subscribeChannel;
+    private MessageHandler messageHandler;
+    // Counter for the number of consumed messages so far to buffer up before we
+    // send the Consume message back to the server along with the last/largest
+    // message seq ID seen so far in that batch.
+    private int numConsumedMessagesInBuffer = 0;
+    private MessageSeqId lastMessageSeqId;
+    // Queue used for subscribes when the MessageHandler hasn't been registered
+    // yet but we've already received subscription messages from the server.
+    // This will be lazily created as needed.
+    private Queue<Message> subscribeMsgQueue;
+    // Set to store all of the outstanding subscribed messages that are pending
+    // to be consumed by the client app's MessageHandler. If this ever grows too
+    // big (e.g. problem at the client end for message consumption), we can
+    // throttle things by temporarily setting the Subscribe Netty Channel
+    // to not be readable. When the Set has shrunk sufficiently, we can turn the
+    // channel back on to read new messages.
+    private Set<Message> outstandingMsgSet;
+
+    public SubscribeResponseHandler(ResponseHandler responseHandler) {
+        this.responseHandler = responseHandler;
+    }
+
+    // Public getter to retrieve the original PubSubData used for the Subscribe
+    // request.
+    public PubSubData getOrigSubData() {
+        return origSubData;
+    }
+
+    // Main method to handle Subscribe responses from the server that we sent
+    // a Subscribe Request to.
+    public void handleSubscribeResponse(PubSubResponse response, PubSubData pubSubData, Channel channel)
+            throws Exception {
+        // If this was not a successful response to the Subscribe request, we
+        // won't be using the Netty Channel created so just close it.
+        if (!response.getStatusCode().equals(StatusCode.SUCCESS)) {
+            HedwigClient.getResponseHandlerFromChannel(channel).channelClosedExplicitly = true;
+            channel.close();
+        }
+
+        if (logger.isDebugEnabled())
+            logger.debug("Handling a Subscribe response: " + response + ", pubSubData: " + pubSubData + ", host: "
+                    + HedwigClient.getHostFromChannel(channel));
+        switch (response.getStatusCode()) {
+        case SUCCESS:
+            // For successful Subscribe requests, store this Channel locally
+            // and set it to not be readable initially.
+            // This way we won't be delivering messages for this topic
+            // subscription until the client explicitly says so.
+            subscribeChannel = channel;
+            subscribeChannel.setReadable(false);
+            // Store the original PubSubData used to create this successful
+            // Subscribe request.
+            origSubData = pubSubData;
+            // Store the mapping for the TopicSubscriber to the Channel.
+            // This is so we can control the starting and stopping of
+            // message deliveries from the server on that Channel. Store
+            // this only on a successful ack response from the server.
+            TopicSubscriber topicSubscriber = new TopicSubscriber(pubSubData.topic, pubSubData.subscriberId);
+            responseHandler.getSubscriber().setChannelForTopic(topicSubscriber, channel);
+            // Lazily create the Set to keep track of outstanding Messages
+            // to be consumed by the client app. At this stage, delivery for
+            // that topic hasn't started yet so creation of this Set should
+            // be thread safe. We'll create the Set with an initial capacity
+            // equal to the configured parameter for the maximum number of
+            // outstanding messages to allow. The load factor will be set to
+            // 1.0f which means we'll only rehash and allocate more space if
+            // we ever exceed the initial capacity. That should be okay
+            // because when that happens, things are slow already and piling
+            // up on the client app side to consume messages.
+            outstandingMsgSet = new HashSet<Message>(
+                    responseHandler.getConfiguration().getMaximumOutstandingMessages(), 1.0f);
+            // Response was success so invoke the callback's operationFinished
+            // method.
+            pubSubData.callback.operationFinished(pubSubData.context, null);
+            break;
+        case CLIENT_ALREADY_SUBSCRIBED:
+            // For Subscribe requests, the server says that the client is
+            // already subscribed to it.
+            pubSubData.callback.operationFailed(pubSubData.context, new ClientAlreadySubscribedException(
+                    "Client is already subscribed for topic: " + pubSubData.topic.toStringUtf8() + ", subscriberId: "
+                            + pubSubData.subscriberId.toStringUtf8()));
+            break;
+        case SERVICE_DOWN:
+            // Response was service down failure so just invoke the callback's
+            // operationFailed method.
+            pubSubData.callback.operationFailed(pubSubData.context, new ServiceDownException(
+                    "Server responded with a SERVICE_DOWN status"));
+            break;
+        case NOT_RESPONSIBLE_FOR_TOPIC:
+            // Redirect response so we'll need to repost the original Subscribe
+            // Request
+            responseHandler.handleRedirectResponse(response, pubSubData, channel);
+            break;
+        default:
+            // Consider all other status codes as errors, operation failed
+            // cases.
+            logger.error("Unexpected error response from server for PubSubResponse: " + response);
+            pubSubData.callback.operationFailed(pubSubData.context, new ServiceDownException(
+                    "Server responded with a status code of: " + response.getStatusCode()));
+            break;
+        }
+    }
+
+    // Main method to handle consuming a message for a topic that the client is
+    // subscribed to.
+    public void handleSubscribeMessage(PubSubResponse response) {
+        if (logger.isDebugEnabled())
+            logger.debug("Handling a Subscribe message in response: " + response + ", topic: "
+                    + origSubData.topic.toStringUtf8() + ", subscriberId: " + origSubData.subscriberId.toStringUtf8());
+        Message message = response.getMessage();
+        // Consume the message asynchronously that the client is subscribed
+        // to. Do this only if delivery for the subscription has started and
+        // a MessageHandler has been registered for the TopicSubscriber.
+        if (messageHandler != null) {
+            asyncMessageConsume(message);
+        } else {
+            // MessageHandler has not yet been registered so queue up these
+            // messages for the Topic Subscription. Make the initial lazy
+            // creation of the message queue thread safe just so we don't
+            // run into a race condition where two simultaneous threads process
+            // a received message and both try to create a new instance of
+            // the message queue. Performance overhead should be okay
+            // because the delivery of the topic has not even started yet
+            // so these messages are not consumed and just buffered up here.
+            synchronized (this) {
+                if (subscribeMsgQueue == null)
+                    subscribeMsgQueue = new LinkedList<Message>();
+            }
+            if (logger.isDebugEnabled())
+                logger
+                        .debug("Message has arrived but Subscribe channel does not have a registered MessageHandler yet so queueing up the message: "
+                                + message);
+            subscribeMsgQueue.add(message);
+        }
+    }
+
+    /**
+     * Method called when a message arrives for a subscribe Channel and we want
+     * to consume it asynchronously via the registered MessageHandler (should
+     * not be null when called here).
+     * 
+     * @param message
+     *            Message from Subscribe Channel we want to consume.
+     */
+    protected void asyncMessageConsume(Message message) {
+        if (logger.isDebugEnabled())
+            logger.debug("Call the client app's MessageHandler asynchronously to consume the message: " + message
+                    + ", topic: " + origSubData.topic.toStringUtf8() + ", subscriberId: "
+                    + origSubData.subscriberId.toStringUtf8());
+        // Add this "pending to be consumed" message to the outstandingMsgSet.
+        outstandingMsgSet.add(message);
+        // Check if we've exceeded the max size for the outstanding message set.
+        if (outstandingMsgSet.size() >= responseHandler.getConfiguration().getMaximumOutstandingMessages()
+                && subscribeChannel.isReadable()) {
+            // Too many outstanding messages so throttle it by setting the Netty
+            // Channel to not be readable.
+            if (logger.isDebugEnabled())
+                logger.debug("Too many outstanding messages (" + outstandingMsgSet.size()
+                        + ") so throttling the subscribe netty Channel");
+            subscribeChannel.setReadable(false);
+        }
+        MessageConsumeData messageConsumeData = new MessageConsumeData(origSubData.topic, origSubData.subscriberId,
+                message);
+        messageHandler.consume(origSubData.topic, origSubData.subscriberId, message, responseHandler.getClient()
+                .getConsumeCallback(), messageConsumeData);
+    }
+
+    /**
+     * Method called when the client app's MessageHandler has asynchronously
+     * completed consuming a subscribed message sent from the server. The
+     * contract with the client app is that messages sent to the handler to be
+     * consumed will have the callback response done in the same order. So if we
+     * asynchronously call the MessageHandler to consume messages #1-5, that
+     * should call the messageConsumed method here via the VoidCallback in the
+     * same order. To make this thread safe, since multiple outstanding messages
+     * could be consumed by the client app and then called back to here, make
+     * this method synchronized.
+     * 
+     * @param message
+     *            Message sent from server for topic subscription that has been
+     *            consumed by the client.
+     */
+    protected synchronized void messageConsumed(Message message) {
+        if (logger.isDebugEnabled())
+            logger.debug("Message has been successfully consumed by the client app for message: " + message
+                    + ", topic: " + origSubData.topic.toStringUtf8() + ", subscriberId: "
+                    + origSubData.subscriberId.toStringUtf8());
+        // Update the consumed messages buffer variables
+        if (responseHandler.getConfiguration().isAutoSendConsumeMessageEnabled()) {
+            // Update these variables only if we are auto-sending consume
+            // messages to the server. Otherwise the onus is on the client app
+            // to call the Subscriber consume API to let the server know which
+            // messages it has successfully consumed.
+            numConsumedMessagesInBuffer++;
+            lastMessageSeqId = message.getMsgId();
+        }
+        // Remove this consumed message from the outstanding Message Set.
+        outstandingMsgSet.remove(message);
+
+        // For consume response to server, there is a config param on how many
+        // messages to consume and buffer up before sending the consume request.
+        // We just need to keep a count of the number of messages consumed
+        // and the largest/latest msg ID seen so far in this batch. Messages
+        // should be delivered in order and without gaps. Do this only if
+        // auto-sending of consume messages is enabled.
+        if (responseHandler.getConfiguration().isAutoSendConsumeMessageEnabled()
+                && numConsumedMessagesInBuffer >= responseHandler.getConfiguration().getConsumedMessagesBufferSize()) {
+            // Send the consume request and reset the consumed messages buffer
+            // variables. We will use the same Channel created from the
+            // subscribe request for the TopicSubscriber.
+            if (logger.isDebugEnabled())
+                logger
+                        .debug("Consumed message buffer limit reached so send the Consume Request to the server with lastMessageSeqId: "
+                                + lastMessageSeqId);
+            responseHandler.getSubscriber().doConsume(origSubData, subscribeChannel, lastMessageSeqId);
+            numConsumedMessagesInBuffer = 0;
+            lastMessageSeqId = null;
+        }
+
+        // Check if we throttled message consumption previously when the
+        // outstanding message limit was reached. For now, only turn the
+        // delivery back on if there are no more outstanding messages to
+        // consume. We could make this a configurable parameter if needed.
+        if (!subscribeChannel.isReadable() && outstandingMsgSet.size() == 0) {
+            if (logger.isDebugEnabled())
+                logger
+                        .debug("Message consumption has caught up so okay to turn off throttling of messages on the subscribe channel for topic: "
+                                + origSubData.topic.toStringUtf8()
+                                + ", subscriberId: "
+                                + origSubData.subscriberId.toStringUtf8());
+            subscribeChannel.setReadable(true);
+        }
+    }
+
+    /**
+     * Setter used for Subscribe flows when delivery for the subscription is
+     * started. This is used to register the MessageHandler needed to consumer
+     * the subscribed messages for the topic.
+     * 
+     * @param messageHandler
+     *            MessageHandler to register for this ResponseHandler instance.
+     */
+    public void setMessageHandler(MessageHandler messageHandler) {
+        if (logger.isDebugEnabled())
+            logger.debug("Setting the messageHandler for topic: " + origSubData.topic.toStringUtf8()
+                    + ", subscriberId: " + origSubData.subscriberId.toStringUtf8());
+        this.messageHandler = messageHandler;
+        // Once the MessageHandler is registered, see if we have any queued up
+        // subscription messages sent to us already from the server. If so,
+        // consume those first. Do this only if the MessageHandler registered is
+        // not null (since that would be the HedwigSubscriber.stopDelivery
+        // call).
+        if (messageHandler != null && subscribeMsgQueue != null && subscribeMsgQueue.size() > 0) {
+            if (logger.isDebugEnabled())
+                logger.debug("Consuming " + subscribeMsgQueue.size() + " queued up messages for topic: "
+                        + origSubData.topic.toStringUtf8() + ", subscriberId: "
+                        + origSubData.subscriberId.toStringUtf8());
+            for (Message message : subscribeMsgQueue) {
+                asyncMessageConsume(message);
+            }
+            // Now we can remove the queued up messages since they are all
+            // consumed.
+            subscribeMsgQueue.clear();
+        }
+    }
+
+    /**
+     * Getter for the MessageHandler that is set for this subscribe channel.
+     * 
+     * @return The MessageHandler for consuming messages
+     */
+    public MessageHandler getMessageHandler() {
+        return messageHandler;
+    }
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/UnsubscribeResponseHandler.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/UnsubscribeResponseHandler.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/UnsubscribeResponseHandler.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/handlers/UnsubscribeResponseHandler.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,83 @@
+/**
+ * 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.hedwig.client.handlers;
+
+import org.apache.log4j.Logger;
+import org.jboss.netty.channel.Channel;
+
+import org.apache.hedwig.client.data.PubSubData;
+import org.apache.hedwig.client.netty.HedwigClient;
+import org.apache.hedwig.client.netty.ResponseHandler;
+import org.apache.hedwig.exceptions.PubSubException.ClientNotSubscribedException;
+import org.apache.hedwig.exceptions.PubSubException.ServiceDownException;
+import org.apache.hedwig.protocol.PubSubProtocol.PubSubResponse;
+
+public class UnsubscribeResponseHandler {
+
+    private static Logger logger = Logger.getLogger(UnsubscribeResponseHandler.class);
+
+    private final ResponseHandler responseHandler;
+
+    public UnsubscribeResponseHandler(ResponseHandler responseHandler) {
+        this.responseHandler = responseHandler;
+    }
+
+    // Main method to handle Unsubscribe Response messages from the server.
+    public void handleUnsubscribeResponse(PubSubResponse response, PubSubData pubSubData, Channel channel)
+            throws Exception {
+        if (logger.isDebugEnabled())
+            logger.debug("Handling an Unsubscribe response: " + response + ", pubSubData: " + pubSubData + ", host: "
+                    + HedwigClient.getHostFromChannel(channel));
+        switch (response.getStatusCode()) {
+        case SUCCESS:
+            // For successful Unsubscribe requests, we can now safely close the
+            // Subscribe Channel and any cached data for that TopicSubscriber.
+            responseHandler.getSubscriber().closeSubscription(pubSubData.topic, pubSubData.subscriberId);
+            // Response was success so invoke the callback's operationFinished
+            // method.
+            pubSubData.callback.operationFinished(pubSubData.context, null);
+            break;
+        case CLIENT_NOT_SUBSCRIBED:
+            // For Unsubscribe requests, the server says that the client was
+            // never subscribed to the topic.
+            pubSubData.callback.operationFailed(pubSubData.context, new ClientNotSubscribedException(
+                    "Client was never subscribed to topic: " + pubSubData.topic.toStringUtf8() + ", subscriberId: "
+                            + pubSubData.subscriberId.toStringUtf8()));
+            break;
+        case SERVICE_DOWN:
+            // Response was service down failure so just invoke the callback's
+            // operationFailed method.
+            pubSubData.callback.operationFailed(pubSubData.context, new ServiceDownException(
+                    "Server responded with a SERVICE_DOWN status"));
+            break;
+        case NOT_RESPONSIBLE_FOR_TOPIC:
+            // Redirect response so we'll need to repost the original
+            // Unsubscribe Request
+            responseHandler.handleRedirectResponse(response, pubSubData, channel);
+            break;
+        default:
+            // Consider all other status codes as errors, operation failed
+            // cases.
+            logger.error("Unexpected error response from server for PubSubResponse: " + response);
+            pubSubData.callback.operationFailed(pubSubData.context, new ServiceDownException(
+                    "Server responded with a status code of: " + response.getStatusCode()));
+            break;
+        }
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ClientChannelPipelineFactory.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ClientChannelPipelineFactory.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ClientChannelPipelineFactory.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ClientChannelPipelineFactory.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,58 @@
+/**
+ * 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.hedwig.client.netty;
+
+import org.jboss.netty.channel.ChannelPipeline;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
+import org.jboss.netty.handler.codec.frame.LengthFieldPrepender;
+import org.jboss.netty.handler.codec.protobuf.ProtobufDecoder;
+import org.jboss.netty.handler.codec.protobuf.ProtobufEncoder;
+import org.jboss.netty.handler.ssl.SslHandler;
+
+import org.apache.hedwig.protocol.PubSubProtocol;
+
+public class ClientChannelPipelineFactory implements ChannelPipelineFactory {
+
+    private HedwigClient client;
+
+    public ClientChannelPipelineFactory(HedwigClient client) {
+        this.client = client;
+    }
+
+    // Retrieve a ChannelPipeline from the factory.
+    public ChannelPipeline getPipeline() throws Exception {
+        // Create a new ChannelPipline using the factory method from the
+        // Channels helper class.
+        ChannelPipeline pipeline = Channels.pipeline();        
+        if (client.getSslFactory() != null) {
+            pipeline.addLast("ssl", new SslHandler(client.getSslFactory().getEngine()));
+        }        
+        pipeline.addLast("lengthbaseddecoder", new LengthFieldBasedFrameDecoder(client.getConfiguration()
+                .getMaximumMessageSize(), 0, 4, 0, 4));
+        pipeline.addLast("lengthprepender", new LengthFieldPrepender(4));
+
+        pipeline.addLast("protobufdecoder", new ProtobufDecoder(PubSubProtocol.PubSubResponse.getDefaultInstance()));
+        pipeline.addLast("protobufencoder", new ProtobufEncoder());
+
+        pipeline.addLast("responsehandler", new ResponseHandler(client));
+        return pipeline;
+    }
+
+}

Added: hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ConnectCallback.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ConnectCallback.java?rev=987314&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ConnectCallback.java (added)
+++ hadoop/zookeeper/trunk/src/contrib/hedwig/client/src/main/java/org/apache/hedwig/client/netty/ConnectCallback.java Thu Aug 19 21:25:13 2010
@@ -0,0 +1,122 @@
+/**
+ * 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.hedwig.client.netty;
+
+import java.net.InetSocketAddress;
+import java.util.LinkedList;
+
+import org.apache.log4j.Logger;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+
+import com.google.protobuf.ByteString;
+import org.apache.hedwig.client.conf.ClientConfiguration;
+import org.apache.hedwig.client.data.PubSubData;
+import org.apache.hedwig.exceptions.PubSubException.CouldNotConnectException;
+import org.apache.hedwig.protocol.PubSubProtocol.OperationType;
+import org.apache.hedwig.util.HedwigSocketAddress;
+
+public class ConnectCallback implements ChannelFutureListener {
+
+    private static Logger logger = Logger.getLogger(ConnectCallback.class);
+
+    // Private member variables
+    private PubSubData pubSubData;
+    private InetSocketAddress host;
+    private final HedwigClient client;
+    private final HedwigPublisher pub;
+    private final HedwigSubscriber sub;
+    private final ClientConfiguration cfg;
+
+    // Constructor
+    public ConnectCallback(PubSubData pubSubData, InetSocketAddress host, HedwigClient client) {
+        super();
+        this.pubSubData = pubSubData;
+        this.host = host;
+        this.client = client;
+        this.pub = client.getPublisher();
+        this.sub = client.getSubscriber();
+        this.cfg = client.getConfiguration();
+    }
+
+    public void operationComplete(ChannelFuture future) throws Exception {
+        // If the client has stopped, there is no need to proceed with any
+        // callback logic here.
+        if (client.hasStopped())
+            return;
+
+        // Check if the connection to the server was done successfully.
+        if (!future.isSuccess()) {
+            logger.error("Error connecting to host: " + host);
+            // If we were not able to connect to the host, it could be down.
+            ByteString hostString = ByteString.copyFromUtf8(HedwigSocketAddress.sockAddrStr(host));
+            if (pubSubData.connectFailedServers != null && pubSubData.connectFailedServers.contains(hostString)) {
+                // We've already tried to connect to this host before so just
+                // invoke the operationFailed callback.
+                logger.error("Error connecting to host more than once so just invoke the operationFailed callback!");
+                pubSubData.callback.operationFailed(pubSubData.context, new CouldNotConnectException(
+                        "Could not connect to host: " + host));
+            } else {
+                if (logger.isDebugEnabled())
+                    logger.debug("Try to connect to server: " + host + " again for pubSubData: " + pubSubData);
+                // Keep track of this current server that we failed to connect
+                // to but retry the request on the default server host/VIP.
+                // The topic2Host mapping might need to be updated.
+                if (pubSubData.connectFailedServers == null)
+                    pubSubData.connectFailedServers = new LinkedList<ByteString>();
+                pubSubData.connectFailedServers.add(hostString);
+                client.doConnect(pubSubData, cfg.getDefaultServerHost());
+            }
+            // Finished with failure logic so just return.
+            return;
+        }
+
+        // Now that we have connected successfully to the server, see what type
+        // of PubSub request this was.
+        if (logger.isDebugEnabled())
+            logger.debug("Connection to host: " + host + " was successful for pubSubData: " + pubSubData);
+        if (pubSubData.operationType.equals(OperationType.PUBLISH)) {
+            // Publish Request so store this Channel connection in the
+            // HedwigPublisher Map (if it doesn't exist yet) and then
+            // do the publish on the cached channel mapped to the host.
+            // Note that due to race concurrency situations, it is
+            // possible that the cached channel is not the same one
+            // as the channel established here. If that is the case,
+            // this channel will be closed but we'll always publish on the
+            // cached channel in the HedwigPublisher.host2Channel map.
+            pub.storeHost2ChannelMapping(future.getChannel());
+            pub.doPublish(pubSubData, pub.host2Channel.get(HedwigClient.getHostFromChannel(future.getChannel())));
+        } else if (pubSubData.operationType.equals(OperationType.UNSUBSCRIBE)) {
+            // Unsubscribe Request so store this Channel connection in the
+            // HedwigPublisher Map (if it doesn't exist yet) and then do the
+            // unsubscribe. Unsubscribe requests will share and reuse
+            // the netty Channel connections that Publish requests use.
+            pub.storeHost2ChannelMapping(future.getChannel());
+            sub.doSubUnsub(pubSubData, pub.host2Channel.get(HedwigClient.getHostFromChannel(future.getChannel())));
+        } else {
+            // Subscribe Request. We do not store the Channel connection yet for
+            // Subscribes here. This will be done only when we've found the
+            // right server topic master. That is only determined when we
+            // receive a successful server ack response to the Subscribe
+            // request (handled in ResponseHandler). There is no need to store
+            // the Unsubscribe channel connection as we won't use it again.
+            sub.doSubUnsub(pubSubData, future.getChannel());
+        }
+    }
+
+}



Mime
View raw message