Creating a branch for experimenting with some ideas for JMS client.rajith_jms_client

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/rajith_jms_client@1128369 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 160 insertions, 0 deletions

diff --git a/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h b/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
new file mode 100644
index 0000000000..c10c77ae18
--- /dev/null
+++ b/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
@@ -0,0 +1,160 @@
+#ifndef QPID_CLIENT_AMQP0_10_SENDERIMPL_H
+#define QPID_CLIENT_AMQP0_10_SENDERIMPL_H
+
+/*
+ *
+ * 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.
+ *
+ */
+#include "qpid/messaging/Address.h"
+#include "qpid/messaging/Message.h"
+#include "qpid/messaging/SenderImpl.h"
+#include "qpid/client/AsyncSession.h"
+#include "qpid/client/amqp0_10/SessionImpl.h"
+#include <memory>
+#include <boost/intrusive_ptr.hpp>
+#include <boost/ptr_container/ptr_deque.hpp>
+
+namespace qpid {
+namespace client {
+namespace amqp0_10 {
+
+class AddressResolution;
+class MessageSink;
+struct OutgoingMessage;
+
+/**
+ *
+ */
+class SenderImpl : public qpid::messaging::SenderImpl
+{
+  public:
+    enum State {UNRESOLVED, ACTIVE, CANCELLED};
+
+    SenderImpl(SessionImpl& parent, const std::string& name, 
+               const qpid::messaging::Address& address);
+    void send(const qpid::messaging::Message&, bool sync);
+    void close();
+    void setCapacity(uint32_t);
+    uint32_t getCapacity();
+    uint32_t getUnsettled();
+    void init(qpid::client::AsyncSession, AddressResolution&);
+    const std::string& getName() const;
+    qpid::messaging::Session getSession() const;
+
+  private:
+    mutable sys::Mutex lock;
+    boost::intrusive_ptr<SessionImpl> parent;
+    const std::string name;
+    const qpid::messaging::Address address;
+    State state;
+    std::auto_ptr<MessageSink> sink;
+
+    qpid::client::AsyncSession session;
+    std::string destination;
+    std::string routingKey;
+
+    typedef boost::ptr_deque<OutgoingMessage> OutgoingMessages;
+    OutgoingMessages outgoing;
+    uint32_t capacity;
+    uint32_t window;
+    bool flushed;
+    const bool unreliable;
+
+    uint32_t checkPendingSends(bool flush);
+    // Dummy ScopedLock parameter means call with lock held
+    uint32_t checkPendingSends(bool flush, const sys::Mutex::ScopedLock&);
+    void replay(const sys::Mutex::ScopedLock&); 
+    void waitForCapacity();
+
+    //logic for application visible methods:
+    void sendImpl(const qpid::messaging::Message&);
+    void sendUnreliable(const qpid::messaging::Message&);
+    void closeImpl();
+
+
+    //functors for application visible methods (allowing locking and
+    //retry to be centralised):
+    struct Command
+    {
+        SenderImpl& impl;
+
+        Command(SenderImpl& i) : impl(i) {}
+    };
+
+    struct Send : Command
+    {
+        const qpid::messaging::Message* message;
+        bool repeat;
+
+        Send(SenderImpl& i, const qpid::messaging::Message* m) : Command(i), message(m), repeat(true) {}
+        void operator()() 
+        {
+            impl.waitForCapacity();
+            //from this point message will be recorded if there is any
+            //failure (and replayed) so need not repeat the call
+            repeat = false;
+            impl.sendImpl(*message);
+        }
+    };
+
+    struct UnreliableSend : Command
+    {
+        const qpid::messaging::Message* message;
+
+        UnreliableSend(SenderImpl& i, const qpid::messaging::Message* m) : Command(i), message(m) {}
+        void operator()() 
+        {
+            //TODO: ideally want to put messages on the outbound
+            //queue and pull them off in io thread, but the old
+            //0-10 client doesn't support that option so for now
+            //we simply don't queue unreliable messages
+            impl.sendUnreliable(*message);                
+        }
+    };
+
+    struct Close : Command
+    {
+        Close(SenderImpl& i) : Command(i) {}
+        void operator()() { impl.closeImpl(); }
+    };
+
+    struct CheckPendingSends : Command
+    {
+        bool flush;
+        uint32_t pending;
+        CheckPendingSends(SenderImpl& i, bool f) : Command(i), flush(f), pending(0) {}
+        void operator()() { pending = impl.checkPendingSends(flush); }
+    };
+
+    //helper templates for some common patterns
+    template <class F> void execute()
+    {
+        F f(*this);
+        parent->execute(f);
+    }
+    
+    template <class F, class P> bool execute1(P p)
+    {
+        F f(*this, p);
+        return parent->execute(f);
+    }    
+};
+}}} // namespace qpid::client::amqp0_10
+
+#endif  /*!QPID_CLIENT_AMQP0_10_SENDERIMPL_H*/