You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2012/10/19 19:16:10 UTC
svn commit: r1400178 [5/5] - in /qpid/trunk/qpid/cpp: ./ src/ src/qpid/amqp/
src/qpid/broker/ src/qpid/broker/amqp/ src/qpid/messaging/
src/qpid/messaging/amqp/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,97 @@
+/*
+ *
+ * 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 "ReceiverContext.h"
+#include "qpid/messaging/Duration.h"
+#include "qpid/messaging/Message.h"
+extern "C" {
+#include <proton/engine.h>
+}
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+//TODO: proper conversion to wide string for address
+ReceiverContext::ReceiverContext(pn_session_t* session, const std::string& n, const std::string& s)
+ : name(n),
+ source(s),
+ receiver(pn_receiver(session, source.c_str())),
+ capacity(0) {}
+ReceiverContext::~ReceiverContext()
+{
+ pn_link_free(receiver);
+}
+
+void ReceiverContext::setCapacity(uint32_t c)
+{
+ if (c != capacity) {
+ //stop
+ capacity = c;
+ //reissue credit
+ }
+}
+
+uint32_t ReceiverContext::getCapacity()
+{
+ return capacity;
+}
+
+uint32_t ReceiverContext::getAvailable()
+{
+ uint32_t count(0);
+ for (pn_delivery_t* d = pn_unsettled_head(receiver); d; d = pn_unsettled_next(d)) {
+ ++count;
+ if (d == pn_link_current(receiver)) break;
+ }
+ return count;
+}
+
+uint32_t ReceiverContext::getUnsettled()
+{
+ uint32_t count(0);
+ for (pn_delivery_t* d = pn_unsettled_head(receiver); d; d = pn_unsettled_next(d)) {
+ ++count;
+ }
+ return count;
+}
+
+void ReceiverContext::close()
+{
+
+}
+
+const std::string& ReceiverContext::getName() const
+{
+ return name;
+}
+
+const std::string& ReceiverContext::getSource() const
+{
+ return source;
+}
+
+bool ReceiverContext::isClosed() const
+{
+ return false;//TODO
+}
+
+
+
+}}} // namespace qpid::messaging::amqp
Copied: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h (from r1400177, qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h?p2=qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h&p1=qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h&r1=1400177&r2=1400178&rev=1400178&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverContext.h Fri Oct 19 17:16:07 2012
@@ -1,5 +1,5 @@
-#ifndef QPID_MESSAGING_SENDERIMPL_H
-#define QPID_MESSAGING_SENDERIMPL_H
+#ifndef QPID_MESSAGING_AMQP_RECEIVERCONTEXT_H
+#define QPID_MESSAGING_AMQP_RECEIVERCONTEXT_H
/*
*
@@ -10,9 +10,9 @@
* 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
@@ -21,27 +21,43 @@
* under the License.
*
*/
-#include "qpid/RefCounted.h"
+#include <string>
+#include "qpid/sys/IntegerTypes.h"
+
+struct pn_link_t;
+struct pn_session_t;
namespace qpid {
namespace messaging {
+class Duration;
class Message;
-class Session;
-class SenderImpl : public virtual qpid::RefCounted
+namespace amqp {
+
+/**
+ *
+ */
+class ReceiverContext
{
public:
- virtual ~SenderImpl() {}
- virtual void send(const Message& message, bool sync) = 0;
- virtual void close() = 0;
- virtual void setCapacity(uint32_t) = 0;
- virtual uint32_t getCapacity() = 0;
- virtual uint32_t getUnsettled() = 0;
- virtual const std::string& getName() const = 0;
- virtual Session getSession() const = 0;
+ ReceiverContext(pn_session_t* session, const std::string& name, const std::string& source);
+ ~ReceiverContext();
+ void setCapacity(uint32_t);
+ uint32_t getCapacity();
+ uint32_t getAvailable();
+ uint32_t getUnsettled();
+ void close();
+ const std::string& getName() const;
+ const std::string& getSource() const;
+ bool isClosed() const;
private:
+ friend class ConnectionContext;
+ const std::string name;
+ const std::string source;
+ pn_link_t* receiver;
+ uint32_t capacity;
};
-}} // namespace qpid::messaging
+}}} // namespace qpid::messaging::amqp
-#endif /*!QPID_MESSAGING_SENDERIMPL_H*/
+#endif /*!QPID_MESSAGING_AMQP_RECEIVERCONTEXT_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,106 @@
+/*
+ *
+ * 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 "ReceiverHandle.h"
+#include "ConnectionContext.h"
+#include "SessionContext.h"
+#include "SessionHandle.h"
+#include "ReceiverContext.h"
+#include "qpid/messaging/Duration.h"
+#include "qpid/messaging/exceptions.h"
+#include "qpid/messaging/Message.h"
+#include "qpid/messaging/Session.h"
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+ReceiverHandle::ReceiverHandle(boost::shared_ptr<ConnectionContext> c,
+ boost::shared_ptr<SessionContext> s,
+ boost::shared_ptr<ReceiverContext> r
+) : connection(c), session(s), receiver(r) {}
+
+
+bool ReceiverHandle::get(qpid::messaging::Message& message, qpid::messaging::Duration timeout)
+{
+ return connection->get(session, receiver, message, timeout);
+}
+
+qpid::messaging::Message ReceiverHandle::get(qpid::messaging::Duration timeout)
+{
+ qpid::messaging::Message result;
+ if (!get(result, timeout)) throw qpid::messaging::NoMessageAvailable();
+ return result;
+}
+
+bool ReceiverHandle::fetch(qpid::messaging::Message& message, qpid::messaging::Duration timeout)
+{
+ return connection->fetch(session, receiver, message, timeout);
+}
+
+qpid::messaging::Message ReceiverHandle::fetch(qpid::messaging::Duration timeout)
+{
+ qpid::messaging::Message result;
+ if (!fetch(result, timeout)) throw qpid::messaging::NoMessageAvailable();
+ return result;
+}
+
+void ReceiverHandle::setCapacity(uint32_t capacity)
+{
+ connection->setCapacity(receiver, capacity);
+}
+
+uint32_t ReceiverHandle::getCapacity()
+{
+ return connection->getCapacity(receiver);
+}
+
+uint32_t ReceiverHandle::getAvailable()
+{
+ return connection->getAvailable(receiver);
+}
+
+uint32_t ReceiverHandle::getUnsettled()
+{
+ return connection->getUnsettled(receiver);
+}
+
+void ReceiverHandle::close()
+{
+ session->closeReceiver(getName());
+}
+
+const std::string& ReceiverHandle::getName() const
+{
+ return receiver->getName();
+}
+
+qpid::messaging::Session ReceiverHandle::getSession() const
+{
+ //create new SessionHandle instance; i.e. create new handle that shares the same context
+ return qpid::messaging::Session(new SessionHandle(connection, session));
+}
+
+bool ReceiverHandle::isClosed() const
+{
+ return receiver->isClosed();
+}
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/ReceiverHandle.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,63 @@
+#ifndef QPID_MESSAGING_AMQP_RECEIVERHANDLE_H
+#define QPID_MESSAGING_AMQP_RECEIVERHANDLE_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 <boost/shared_ptr.hpp>
+#include "qpid/messaging/ReceiverImpl.h"
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+class ConnectionContext;
+class SessionContext;
+class ReceiverContext;
+/**
+ *
+ */
+class ReceiverHandle : public qpid::messaging::ReceiverImpl
+{
+ public:
+ ReceiverHandle(boost::shared_ptr<ConnectionContext>,
+ boost::shared_ptr<SessionContext>,
+ boost::shared_ptr<ReceiverContext>
+ );
+ bool get(Message& message, qpid::messaging::Duration timeout);
+ qpid::messaging::Message get(qpid::messaging::Duration timeout);
+ bool fetch(Message& message, qpid::messaging::Duration timeout);
+ qpid::messaging::Message fetch(qpid::messaging::Duration timeout);
+ void setCapacity(uint32_t);
+ uint32_t getCapacity();
+ uint32_t getAvailable();
+ uint32_t getUnsettled();
+ void close();
+ const std::string& getName() const;
+ qpid::messaging::Session getSession() const;
+ bool isClosed() const;
+ private:
+ boost::shared_ptr<ConnectionContext> connection;
+ boost::shared_ptr<SessionContext> session;
+ boost::shared_ptr<ReceiverContext> receiver;
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_RECEIVERHANDLE_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,162 @@
+/*
+ *
+ * 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 "ConnectionContext.h"
+#include "qpid/messaging/amqp/Sasl.h"
+#include "qpid/messaging/exceptions.h"
+#include "qpid/sys/SecurityLayer.h"
+#include "qpid/log/Statement.h"
+#include "qpid/Sasl.h"
+#include "qpid/SaslFactory.h"
+#include "qpid/StringUtils.h"
+#include <sstream>
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+Sasl::Sasl(const std::string& id, ConnectionContext& c, const std::string& hostname_)
+ : qpid::amqp::SaslClient(id), context(c),
+ sasl(qpid::SaslFactory::getInstance().create(c.username, c.password, c.service, hostname_, c.minSsf, c.maxSsf, false)),
+ hostname(hostname_), readHeader(true), writeHeader(true), haveOutput(false), state(NONE) {}
+
+std::size_t Sasl::decode(const char* buffer, std::size_t size)
+{
+ size_t decoded = 0;
+ if (readHeader) {
+ decoded += readProtocolHeader(buffer, size);
+ readHeader = !decoded;
+ }
+ if (state == NONE && decoded < size) {
+ decoded += read(buffer + decoded, size - decoded);
+ }
+ QPID_LOG(trace, id << " Sasl::decode(" << size << "): " << decoded);
+ return decoded;
+}
+
+std::size_t Sasl::encode(char* buffer, std::size_t size)
+{
+ size_t encoded = 0;
+ if (writeHeader) {
+ encoded += writeProtocolHeader(buffer, size);
+ writeHeader = !encoded;
+ }
+ if (state == NONE && encoded < size) {
+ encoded += write(buffer + encoded, size - encoded);
+ }
+ haveOutput = (encoded == size);
+ QPID_LOG(trace, id << " Sasl::encode(" << size << "): " << encoded);
+ return encoded;
+}
+
+bool Sasl::canEncode()
+{
+ QPID_LOG(trace, id << " Sasl::canEncode(): " << writeHeader << " || " << haveOutput);
+ return writeHeader || haveOutput;
+}
+
+void Sasl::mechanisms(const std::string& offered)
+{
+ QPID_LOG_CAT(debug, protocol, id << " Received SASL-MECHANISMS(" << offered << ")");
+ std::string response;
+
+ std::string mechanisms;
+ if (context.mechanism.size()) {
+ std::vector<std::string> allowed = split(context.mechanism, " ");
+ std::vector<std::string> supported = split(offered, " ");
+ std::stringstream intersection;
+ for (std::vector<std::string>::const_iterator i = allowed.begin(); i != allowed.end(); ++i) {
+ if (std::find(supported.begin(), supported.end(), *i) != supported.end()) {
+ intersection << *i << " ";
+ }
+ }
+ mechanisms = intersection.str();
+ } else {
+ mechanisms = offered;
+ }
+
+ if (sasl->start(mechanisms, response)) {
+ init(sasl->getMechanism(), &response, hostname.size() ? &hostname : 0);
+ } else {
+ init(sasl->getMechanism(), 0, hostname.size() ? &hostname : 0);
+ }
+ haveOutput = true;
+ context.activateOutput();
+}
+void Sasl::challenge(const std::string& challenge)
+{
+ QPID_LOG_CAT(debug, protocol, id << " Received SASL-CHALLENGE(" << challenge.size() << " bytes)");
+ std::string r = sasl->step(challenge);
+ response(&r);
+ haveOutput = true;
+ context.activateOutput();
+}
+namespace {
+const std::string EMPTY;
+}
+void Sasl::challenge()
+{
+ QPID_LOG_CAT(debug, protocol, id << " Received SASL-CHALLENGE(null)");
+ std::string r = sasl->step(EMPTY);
+ response(&r);
+}
+void Sasl::outcome(uint8_t result, const std::string& extra)
+{
+ QPID_LOG_CAT(debug, protocol, id << " Received SASL-OUTCOME(" << result << ", " << extra << ")");
+ outcome(result);
+}
+void Sasl::outcome(uint8_t result)
+{
+ QPID_LOG_CAT(debug, protocol, id << " Received SASL-OUTCOME(" << result << ")");
+ if (result) state = FAILED;
+ else state = SUCCEEDED;
+
+ securityLayer = sasl->getSecurityLayer(context.maxFrameSize);
+ if (securityLayer.get()) {
+ securityLayer->init(&context);
+ }
+ context.activateOutput();
+}
+
+qpid::sys::Codec* Sasl::getCodec()
+{
+ switch (state) {
+ case SUCCEEDED: return static_cast<qpid::sys::Codec*>(securityLayer.get());
+ case FAILED: throw qpid::messaging::UnauthorizedAccess("Failed to authenticate");
+ case NONE: return static_cast<qpid::sys::Codec*>(this);
+ }
+ return 0;
+}
+
+bool Sasl::authenticated()
+{
+ switch (state) {
+ case SUCCEEDED: return true;
+ case FAILED: throw qpid::messaging::UnauthorizedAccess("Failed to authenticate");
+ case NONE: default: return false;
+ }
+}
+
+std::string Sasl::getAuthenticatedUsername()
+{
+ return sasl->getUserId();
+}
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Sasl.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,72 @@
+#ifndef QPID_MESSAGING_AMQP_SASL_H
+#define QPID_MESSAGING_AMQP_SASL_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/sys/Codec.h"
+#include "qpid/amqp/SaslClient.h"
+#include <memory>
+
+namespace qpid {
+class Sasl;
+namespace sys {
+class SecurityLayer;
+}
+namespace messaging {
+class ConnectionOptions;
+namespace amqp {
+class ConnectionContext;
+
+/**
+ *
+ */
+class Sasl : public qpid::sys::Codec, qpid::amqp::SaslClient
+{
+ public:
+ Sasl(const std::string& id, ConnectionContext& context, const std::string& hostname);
+ std::size_t decode(const char* buffer, std::size_t size);
+ std::size_t encode(char* buffer, std::size_t size);
+ bool canEncode();
+
+ bool authenticated();
+ qpid::sys::Codec* getCodec();
+ std::string getAuthenticatedUsername();
+ private:
+ ConnectionContext& context;
+ std::auto_ptr<qpid::Sasl> sasl;
+ std::string hostname;
+ bool readHeader;
+ bool writeHeader;
+ bool haveOutput;
+ enum {
+ NONE, FAILED, SUCCEEDED
+ } state;
+ std::auto_ptr<qpid::sys::SecurityLayer> securityLayer;
+
+ void mechanisms(const std::string&);
+ void challenge(const std::string&);
+ void challenge(); //null != empty string
+ void outcome(uint8_t result, const std::string&);
+ void outcome(uint8_t result);
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_SASL_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,332 @@
+/*
+ *
+ * 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/amqp/SenderContext.h"
+#include "qpid/messaging/amqp/EncodedMessage.h"
+#include "qpid/amqp/descriptors.h"
+#include "qpid/amqp/MessageEncoder.h"
+#include "qpid/messaging/exceptions.h"
+#include "qpid/messaging/Message.h"
+#include "qpid/messaging/MessageImpl.h"
+#include "qpid/log/Statement.h"
+extern "C" {
+#include <proton/engine.h>
+}
+#include <boost/shared_ptr.hpp>
+#include <string.h>
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+//TODO: proper conversion to wide string for address
+SenderContext::SenderContext(pn_session_t* session, const std::string& n, const std::string& t)
+ : name(n),
+ target(t),
+ sender(pn_sender(session, target.c_str())), capacity(1000) {}
+
+SenderContext::~SenderContext()
+{
+ pn_link_free(sender);
+}
+
+void SenderContext::close()
+{
+
+}
+
+void SenderContext::setCapacity(uint32_t c)
+{
+ if (c < deliveries.size()) throw qpid::messaging::SenderError("Desired capacity is less than unsettled message count!");
+ capacity = c;
+}
+
+uint32_t SenderContext::getCapacity()
+{
+ return capacity;
+}
+
+uint32_t SenderContext::getUnsettled()
+{
+ return processUnsettled();
+}
+
+const std::string& SenderContext::getName() const
+{
+ return name;
+}
+
+const std::string& SenderContext::getTarget() const
+{
+ return target;
+}
+
+SenderContext::Delivery* SenderContext::send(const qpid::messaging::Message& message)
+{
+ if (processUnsettled() < capacity) {
+ deliveries.push_back(Delivery(nextId++));
+ Delivery& delivery = deliveries.back();
+ delivery.encode(MessageImplAccess::get(message));
+ delivery.send(sender);
+ return &delivery;
+ } else {
+ return 0;
+ }
+}
+
+uint32_t SenderContext::processUnsettled()
+{
+ //remove accepted messages from front of deque
+ while (!deliveries.empty() && deliveries.front().accepted()) {
+ deliveries.pop_front();
+ }
+ return deliveries.size();
+}
+namespace {
+class HeaderAdapter : public qpid::amqp::MessageEncoder::Header
+{
+ public:
+ HeaderAdapter(const qpid::messaging::MessageImpl& impl) : msg(impl) {}
+ virtual bool isDurable() const
+ {
+ return msg.isDurable();
+ }
+ virtual uint8_t getPriority() const
+ {
+ return msg.getPriority();
+ }
+ virtual bool hasTtl() const
+ {
+ return msg.getTtl();
+ }
+ virtual uint32_t getTtl() const
+ {
+ return msg.getTtl();
+ }
+ virtual bool isFirstAcquirer() const
+ {
+ return false;
+ }
+ virtual uint32_t getDeliveryCount() const
+ {
+ return msg.isRedelivered() ? 1 : 0;
+ }
+ private:
+ const qpid::messaging::MessageImpl& msg;
+};
+const std::string EMPTY;
+
+class PropertiesAdapter : public qpid::amqp::MessageEncoder::Properties
+{
+ public:
+ PropertiesAdapter(const qpid::messaging::MessageImpl& impl) : msg(impl) {}
+ bool hasMessageId() const
+ {
+ return getMessageId().size();
+ }
+ std::string getMessageId() const
+ {
+ return msg.getMessageId();
+ }
+
+ bool hasUserId() const
+ {
+ return getUserId().size();
+ }
+
+ std::string getUserId() const
+ {
+ return msg.getUserId();
+ }
+
+ bool hasTo() const
+ {
+ return false;//not yet supported
+ }
+
+ std::string getTo() const
+ {
+ return EMPTY;//not yet supported
+ }
+
+ bool hasSubject() const
+ {
+ return getSubject().size();
+ }
+
+ std::string getSubject() const
+ {
+ return msg.getSubject();
+ }
+
+ bool hasReplyTo() const
+ {
+ return msg.getReplyTo();
+ }
+
+ std::string getReplyTo() const
+ {
+ return msg.getReplyTo().str();
+ }
+
+ bool hasCorrelationId() const
+ {
+ return getCorrelationId().size();
+ }
+
+ std::string getCorrelationId() const
+ {
+ return msg.getCorrelationId();
+ }
+
+ bool hasContentType() const
+ {
+ return getContentType().size();
+ }
+
+ std::string getContentType() const
+ {
+ return msg.getContentType();
+ }
+
+ bool hasContentEncoding() const
+ {
+ return false;//not yet supported
+ }
+
+ std::string getContentEncoding() const
+ {
+ return EMPTY;//not yet supported
+ }
+
+ bool hasAbsoluteExpiryTime() const
+ {
+ return false;//not yet supported
+ }
+
+ int64_t getAbsoluteExpiryTime() const
+ {
+ return 0;//not yet supported
+ }
+
+ bool hasCreationTime() const
+ {
+ return false;//not yet supported
+ }
+
+ int64_t getCreationTime() const
+ {
+ return 0;//not yet supported
+ }
+
+ bool hasGroupId() const
+ {
+ return false;//not yet supported
+ }
+
+ std::string getGroupId() const
+ {
+ return EMPTY;//not yet supported
+ }
+
+ bool hasGroupSequence() const
+ {
+ return false;//not yet supported
+ }
+
+ uint32_t getGroupSequence() const
+ {
+ return 0;//not yet supported
+ }
+
+ bool hasReplyToGroupId() const
+ {
+ return false;//not yet supported
+ }
+
+ std::string getReplyToGroupId() const
+ {
+ return EMPTY;//not yet supported
+ }
+ private:
+ const qpid::messaging::MessageImpl& msg;
+};
+}
+
+SenderContext::Delivery::Delivery(int32_t i) : id(i), token(0) {}
+
+void SenderContext::Delivery::encode(const qpid::messaging::MessageImpl& msg)
+{
+ boost::shared_ptr<const EncodedMessage> original = msg.getEncoded();
+
+ if (original) { //still have the content as received, send at least the bare message unaltered
+ //do we need to alter the header? are durable, priority, ttl, first-acquirer, delivery-count different from what was received?
+ if (original->hasHeaderChanged(msg)) {
+ //since as yet have no annotations, just write the revised header then the rest of the message as received
+ encoded.resize(16/*max header size*/ + original->getBareMessage().size);
+ qpid::amqp::MessageEncoder encoder(encoded.getData(), encoded.getSize());
+ HeaderAdapter header(msg);
+ encoder.writeHeader(header);
+ ::memcpy(encoded.getData() + encoder.getPosition(), original->getBareMessage().data, original->getBareMessage().size);
+ } else {
+ //since as yet have no annotations, if the header hasn't
+ //changed and we still have the original bare message, can
+ //send the entire content as is
+ encoded.resize(original->getSize());
+ ::memcpy(encoded.getData(), original->getData(), original->getSize());
+ }
+ } else {
+ HeaderAdapter header(msg);
+ PropertiesAdapter properties(msg);
+ //compute size:
+ encoded.resize(qpid::amqp::MessageEncoder::getEncodedSize(header, properties, msg.getHeaders(), msg.getBytes()));
+ QPID_LOG(debug, "Sending message, buffer is " << encoded.getSize() << " bytes")
+ qpid::amqp::MessageEncoder encoder(encoded.getData(), encoded.getSize());
+ //write header:
+ encoder.writeHeader(header);
+ //write delivery-annotations, write message-annotations (none yet supported)
+ //write properties
+ encoder.writeProperties(properties);
+ //write application-properties
+ encoder.writeApplicationProperties(msg.getHeaders());
+ //write body
+ if (msg.getBytes().size()) encoder.writeBinary(msg.getBytes(), &qpid::amqp::message::DATA);//structured content not yet directly supported
+ if (encoder.getPosition() < encoded.getSize()) {
+ QPID_LOG(debug, "Trimming buffer from " << encoded.getSize() << " to " << encoder.getPosition());
+ encoded.trim(encoder.getPosition());
+ }
+ //write footer (no annotations yet supported)
+ }
+}
+void SenderContext::Delivery::send(pn_link_t* sender)
+{
+ pn_delivery_tag_t tag;
+ tag.size = sizeof(id);
+ tag.bytes = reinterpret_cast<const char*>(&id);
+ token = pn_delivery(sender, tag);
+ pn_link_send(sender, encoded.getData(), encoded.getSize());
+ pn_link_advance(sender);
+}
+
+bool SenderContext::Delivery::accepted()
+{
+ return pn_delivery_remote_state(token) == PN_ACCEPTED;
+}
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderContext.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,84 @@
+#ifndef QPID_MESSAGING_AMQP_SENDERCONTEXT_H
+#define QPID_MESSAGING_AMQP_SENDERCONTEXT_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 <deque>
+#include <string>
+#include <vector>
+#include "qpid/sys/IntegerTypes.h"
+#include "qpid/messaging/amqp/EncodedMessage.h"
+
+struct pn_delivery_t;
+struct pn_link_t;
+struct pn_session_t;
+
+namespace qpid {
+namespace messaging {
+
+class Message;
+class MessageImpl;
+
+namespace amqp {
+/**
+ *
+ */
+class SenderContext
+{
+ public:
+ class Delivery
+ {
+ public:
+ Delivery(int32_t id);
+ void encode(const qpid::messaging::MessageImpl& message);
+ void send(pn_link_t*);
+ bool accepted();
+ private:
+ int32_t id;
+ pn_delivery_t* token;
+ EncodedMessage encoded;
+ };
+
+ SenderContext(pn_session_t* session, const std::string& name, const std::string& target);
+ ~SenderContext();
+ void close();
+ void setCapacity(uint32_t);
+ uint32_t getCapacity();
+ uint32_t getUnsettled();
+ const std::string& getName() const;
+ const std::string& getTarget() const;
+ Delivery* send(const qpid::messaging::Message& message);
+ private:
+ friend class ConnectionContext;
+ typedef std::deque<Delivery> Deliveries;
+
+ const std::string name;
+ const std::string target;
+ pn_link_t* sender;
+ int32_t nextId;
+ Deliveries deliveries;
+ uint32_t capacity;
+
+ uint32_t processUnsettled();
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_SENDERCONTEXT_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,75 @@
+/*
+ *
+ * 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 "SenderHandle.h"
+#include "ConnectionContext.h"
+#include "SessionContext.h"
+#include "SessionHandle.h"
+#include "SenderContext.h"
+#include "qpid/messaging/Duration.h"
+#include "qpid/messaging/exceptions.h"
+#include "qpid/messaging/Message.h"
+#include "qpid/messaging/Session.h"
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+SenderHandle::SenderHandle(boost::shared_ptr<ConnectionContext> c,
+ boost::shared_ptr<SessionContext> s,
+ boost::shared_ptr<SenderContext> sndr
+) : connection(c), session(s), sender(sndr) {}
+
+void SenderHandle::send(const Message& message, bool sync)
+{
+ connection->send(sender, message, sync);
+}
+
+void SenderHandle::close()
+{
+ session->closeSender(getName());
+}
+
+void SenderHandle::setCapacity(uint32_t capacity)
+{
+ connection->setCapacity(sender, capacity);
+}
+
+uint32_t SenderHandle::getCapacity()
+{
+ return connection->getCapacity(sender);
+}
+
+uint32_t SenderHandle::getUnsettled()
+{
+ return connection->getUnsettled(sender);
+}
+
+const std::string& SenderHandle::getName() const
+{
+ return sender->getName();
+}
+
+qpid::messaging::Session SenderHandle::getSession() const
+{
+ return qpid::messaging::Session(new SessionHandle(connection, session));
+}
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SenderHandle.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,58 @@
+#ifndef QPID_MESSAGING_AMQP_SENDERHANDLE_H
+#define QPID_MESSAGING_AMQP_SENDERHANDLE_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 <boost/shared_ptr.hpp>
+#include "qpid/messaging/SenderImpl.h"
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+class ConnectionContext;
+class SessionContext;
+class SenderContext;
+/**
+ *
+ */
+class SenderHandle : public qpid::messaging::SenderImpl
+{
+ public:
+ SenderHandle(boost::shared_ptr<ConnectionContext> connection,
+ boost::shared_ptr<SessionContext> session,
+ boost::shared_ptr<SenderContext> sender
+ );
+ void send(const Message& message, bool sync);
+ void close();
+ void setCapacity(uint32_t);
+ uint32_t getCapacity();
+ uint32_t getUnsettled();
+ const std::string& getName() const;
+ Session getSession() const;
+ private:
+ boost::shared_ptr<ConnectionContext> connection;
+ boost::shared_ptr<SessionContext> session;
+ boost::shared_ptr<SenderContext> sender;
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_SENDERHANDLE_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,147 @@
+/*
+ *
+ * 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 "SessionContext.h"
+#include "SenderContext.h"
+#include "ReceiverContext.h"
+#include <boost/format.hpp>
+#include "qpid/messaging/Address.h"
+#include "qpid/messaging/Duration.h"
+#include "qpid/messaging/exceptions.h"
+#include "qpid/log/Statement.h"
+extern "C" {
+#include <proton/engine.h>
+}
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+SessionContext::SessionContext(pn_connection_t* connection) : session(pn_session(connection)) {}
+SessionContext::~SessionContext()
+{
+ senders.clear(); receivers.clear();
+ pn_session_free(session);
+}
+
+boost::shared_ptr<SenderContext> SessionContext::createSender(const qpid::messaging::Address& address)
+{
+ std::string name = address.getName();
+
+ int count = 1;
+ for (SenderMap::const_iterator i = senders.find(name); i != senders.end(); i = senders.find(name)) {
+ name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
+ }
+ boost::shared_ptr<SenderContext> s(new SenderContext(session, name, address.str()));
+ senders[name] = s;
+ return s;
+}
+
+boost::shared_ptr<ReceiverContext> SessionContext::createReceiver(const qpid::messaging::Address& address)
+{
+ std::string name = address.getName();
+
+ int count = 1;
+ for (ReceiverMap::const_iterator i = receivers.find(name); i != receivers.end(); i = receivers.find(name)) {
+ name = (boost::format("%1%_%2%") % address.getName() % ++count).str();
+ }
+ boost::shared_ptr<ReceiverContext> r(new ReceiverContext(session, name, address.str()));
+ receivers[name] = r;
+ return r;
+}
+
+boost::shared_ptr<SenderContext> SessionContext::getSender(const std::string& name) const
+{
+ SenderMap::const_iterator i = senders.find(name);
+ if (i == senders.end()) {
+ throw qpid::messaging::KeyError(std::string("No such sender") + name);
+ } else {
+ return i->second;
+ }
+}
+
+boost::shared_ptr<ReceiverContext> SessionContext::getReceiver(const std::string& name) const
+{
+ ReceiverMap::const_iterator i = receivers.find(name);
+ if (i == receivers.end()) {
+ throw qpid::messaging::KeyError(std::string("No such receiver") + name);
+ } else {
+ return i->second;
+ }
+}
+
+void SessionContext::closeReceiver(const std::string&)
+{
+
+}
+
+void SessionContext::closeSender(const std::string&)
+{
+
+}
+
+boost::shared_ptr<ReceiverContext> SessionContext::nextReceiver(qpid::messaging::Duration /*timeout*/)
+{
+ return boost::shared_ptr<ReceiverContext>();
+}
+
+uint32_t SessionContext::getReceivable()
+{
+ return 0;//TODO
+}
+
+uint32_t SessionContext::getUnsettledAcks()
+{
+ return 0;//TODO
+}
+
+qpid::framing::SequenceNumber SessionContext::record(pn_delivery_t* delivery)
+{
+ qpid::framing::SequenceNumber id = next++;
+ unacked[id] = delivery;
+ QPID_LOG(debug, "Recorded delivery " << id << " -> " << delivery);
+ return id;
+}
+
+void SessionContext::acknowledge(DeliveryMap::iterator begin, DeliveryMap::iterator end)
+{
+ for (DeliveryMap::iterator i = begin; i != end; ++i) {
+ QPID_LOG(debug, "Setting disposition for delivery " << i->first << " -> " << i->second);
+ pn_delivery_update(i->second, PN_ACCEPTED);
+ pn_delivery_settle(i->second);//TODO: different settlement modes?
+ }
+ unacked.erase(begin, end);
+}
+
+void SessionContext::acknowledge()
+{
+ QPID_LOG(debug, "acknowledging all " << unacked.size() << " messages");
+ acknowledge(unacked.begin(), unacked.end());
+}
+
+void SessionContext::acknowledge(const qpid::framing::SequenceNumber& id, bool cumulative)
+{
+ DeliveryMap::iterator i = unacked.find(id);
+ if (i != unacked.end()) {
+ acknowledge(cumulative ? unacked.begin() : i, ++i);
+ }
+}
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionContext.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,80 @@
+#ifndef QPID_MESSAGING_AMQP_SESSIONCONTEXT_H
+#define QPID_MESSAGING_AMQP_SESSIONCONTEXT_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 <map>
+#include <string>
+#include <boost/shared_ptr.hpp>
+#include "qpid/sys/IntegerTypes.h"
+#include "qpid/framing/SequenceNumber.h"
+
+struct pn_connection_t;
+struct pn_session_t;
+struct pn_delivery_t;
+
+namespace qpid {
+namespace messaging {
+
+class Address;
+class Duration;
+
+namespace amqp {
+
+class ConnectionContext;
+class SenderContext;
+class ReceiverContext;
+/**
+ *
+ */
+class SessionContext
+{
+ public:
+ SessionContext(pn_connection_t*);
+ ~SessionContext();
+ boost::shared_ptr<SenderContext> createSender(const qpid::messaging::Address& address);
+ boost::shared_ptr<ReceiverContext> createReceiver(const qpid::messaging::Address& address);
+ boost::shared_ptr<SenderContext> getSender(const std::string& name) const;
+ boost::shared_ptr<ReceiverContext> getReceiver(const std::string& name) const;
+ void closeReceiver(const std::string&);
+ void closeSender(const std::string&);
+ boost::shared_ptr<ReceiverContext> nextReceiver(qpid::messaging::Duration timeout);
+ uint32_t getReceivable();
+ uint32_t getUnsettledAcks();
+ private:
+ friend class ConnectionContext;
+ typedef std::map<std::string, boost::shared_ptr<SenderContext> > SenderMap;
+ typedef std::map<std::string, boost::shared_ptr<ReceiverContext> > ReceiverMap;
+ typedef std::map<qpid::framing::SequenceNumber, pn_delivery_t*> DeliveryMap;
+ pn_session_t* session;
+ SenderMap senders;
+ ReceiverMap receivers;
+ DeliveryMap unacked;
+ qpid::framing::SequenceNumber next;
+
+ qpid::framing::SequenceNumber record(pn_delivery_t*);
+ void acknowledge();
+ void acknowledge(const qpid::framing::SequenceNumber& id, bool cummulative);
+ void acknowledge(DeliveryMap::iterator begin, DeliveryMap::iterator end);
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_SESSIONCONTEXT_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.cpp Fri Oct 19 17:16:07 2012
@@ -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.
+ *
+ */
+#include "SessionHandle.h"
+#include "ConnectionContext.h"
+#include "ConnectionHandle.h"
+#include "ReceiverContext.h"
+#include "ReceiverHandle.h"
+#include "SenderContext.h"
+#include "SenderHandle.h"
+#include "SessionContext.h"
+#include "qpid/messaging/Connection.h"
+#include "qpid/messaging/Duration.h"
+#include "qpid/messaging/exceptions.h"
+#include "qpid/messaging/Receiver.h"
+#include "qpid/messaging/Sender.h"
+#include "qpid/messaging/Session.h"
+#include "qpid/log/Statement.h"
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+SessionHandle::SessionHandle(boost::shared_ptr<ConnectionContext> c, boost::shared_ptr<SessionContext> s) : connection(c), session(s) {}
+
+void SessionHandle::commit()
+{
+
+}
+
+void SessionHandle::rollback()
+{
+
+}
+
+void SessionHandle::acknowledge(bool /*sync*/)
+{
+ connection->acknowledge(session, 0, false);
+}
+
+void SessionHandle::acknowledge(qpid::messaging::Message& msg, bool cumulative)
+{
+ //TODO: handle cumulative
+ connection->acknowledge(session, &msg, cumulative);
+}
+
+void SessionHandle::reject(qpid::messaging::Message&)
+{
+
+}
+
+void SessionHandle::release(qpid::messaging::Message&)
+{
+
+}
+
+void SessionHandle::close()
+{
+ connection->endSession(session);
+}
+
+void SessionHandle::sync(bool /*block*/)
+{
+
+}
+
+qpid::messaging::Sender SessionHandle::createSender(const qpid::messaging::Address& address)
+{
+ boost::shared_ptr<SenderContext> sender = session->createSender(address);
+ connection->attach(session, sender);
+ return qpid::messaging::Sender(new SenderHandle(connection, session, sender));
+}
+
+qpid::messaging::Receiver SessionHandle::createReceiver(const qpid::messaging::Address& address)
+{
+ boost::shared_ptr<ReceiverContext> receiver = session->createReceiver(address);
+ connection->attach(session, receiver);
+ return qpid::messaging::Receiver(new ReceiverHandle(connection, session, receiver));
+}
+
+bool SessionHandle::nextReceiver(Receiver& receiver, Duration timeout)
+{
+ boost::shared_ptr<ReceiverContext> r = session->nextReceiver(timeout);
+ if (r) {
+ //TODO: cache handles in this case to avoid frequent allocation
+ receiver = qpid::messaging::Receiver(new ReceiverHandle(connection, session, r));
+ return true;
+ } else {
+ return false;
+ }
+}
+
+qpid::messaging::Receiver SessionHandle::nextReceiver(Duration timeout)
+{
+ qpid::messaging::Receiver r;
+ if (nextReceiver(r, timeout)) return r;
+ else throw qpid::messaging::NoMessageAvailable();
+}
+
+uint32_t SessionHandle::getReceivable()
+{
+ return session->getReceivable();
+}
+
+uint32_t SessionHandle::getUnsettledAcks()
+{
+ return session->getUnsettledAcks();
+}
+
+Sender SessionHandle::getSender(const std::string& name) const
+{
+ return qpid::messaging::Sender(new SenderHandle(connection, session, session->getSender(name)));
+}
+
+Receiver SessionHandle::getReceiver(const std::string& name) const
+{
+ return qpid::messaging::Receiver(new ReceiverHandle(connection, session, session->getReceiver(name)));
+}
+
+Connection SessionHandle::getConnection() const
+{
+ return qpid::messaging::Connection(new ConnectionHandle(connection));
+}
+
+void SessionHandle::checkError()
+{
+
+}
+
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SessionHandle.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,64 @@
+#ifndef QPID_MESSAGING_AMQP_SESSIONIMPL_H
+#define QPID_MESSAGING_AMQP_SESSIONIMPL_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 <boost/shared_ptr.hpp>
+#include "qpid/messaging/SessionImpl.h"
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+class ConnectionContext;
+class SessionContext;
+/**
+ *
+ */
+class SessionHandle : public qpid::messaging::SessionImpl
+{
+ public:
+ SessionHandle(boost::shared_ptr<ConnectionContext>, boost::shared_ptr<SessionContext>);
+ void commit();
+ void rollback();
+ void acknowledge(bool sync);
+ void acknowledge(Message&, bool);
+ void reject(Message&);
+ void release(Message&);
+ void close();
+ void sync(bool block);
+ qpid::messaging::Sender createSender(const Address& address);
+ qpid::messaging::Receiver createReceiver(const Address& address);
+ bool nextReceiver(Receiver& receiver, Duration timeout);
+ qpid::messaging::Receiver nextReceiver(Duration timeout);
+ uint32_t getReceivable();
+ uint32_t getUnsettledAcks();
+ qpid::messaging::Sender getSender(const std::string& name) const;
+ qpid::messaging::Receiver getReceiver(const std::string& name) const;
+ qpid::messaging::Connection getConnection() const;
+ void checkError();
+ private:
+ boost::shared_ptr<ConnectionContext> connection;
+ boost::shared_ptr<SessionContext> session;
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_SESSIONIMPL_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,158 @@
+/*
+ *
+ * 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 "SslTransport.h"
+#include "TransportContext.h"
+#include "qpid/sys/ssl/SslIo.h"
+#include "qpid/sys/ConnectionCodec.h"
+#include "qpid/sys/Poller.h"
+#include "qpid/log/Statement.h"
+#include <boost/bind.hpp>
+#include <boost/format.hpp>
+
+using namespace qpid::sys;
+using namespace qpid::sys::ssl;
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+
+// Static constructor which registers connector here
+namespace {
+Transport* create(TransportContext& c, Poller::shared_ptr p)
+{
+ return new SslTransport(c, p);
+}
+
+struct StaticInit
+{
+ StaticInit()
+ {
+ Transport::add("ssl", &create);
+ };
+} init;
+}
+
+
+SslTransport::SslTransport(TransportContext& c, boost::shared_ptr<Poller> p) : context(c), aio(0), poller(p) {}
+
+void SslTransport::connect(const std::string& host, const std::string& port)
+{
+ assert(!aio);
+ try {
+ socket.connect(host, port);
+ connected(socket);
+ } catch (const std::exception& e) {
+ failed(e.what());
+ }
+
+}
+
+void SslTransport::failed(const std::string& msg)
+{
+ QPID_LOG(debug, "Failed to connect: " << msg);
+ socket.close();
+ context.closed();
+}
+
+void SslTransport::connected(const SslSocket&)
+{
+ context.opened();
+ aio = new SslIO(socket,
+ boost::bind(&SslTransport::read, this, _1, _2),
+ boost::bind(&SslTransport::eof, this, _1),
+ boost::bind(&SslTransport::disconnected, this, _1),
+ boost::bind(&SslTransport::socketClosed, this, _1, _2),
+ 0, // nobuffs
+ boost::bind(&SslTransport::write, this, _1));
+ aio->createBuffers(std::numeric_limits<uint16_t>::max());//note: AMQP 1.0 _can_ handle large frame sizes
+ id = boost::str(boost::format("[%1%]") % socket.getFullAddress());
+ aio->start(poller);
+}
+
+void SslTransport::read(SslIO&, SslIO::BufferBase* buffer)
+{
+ int32_t decoded = context.getCodec().decode(buffer->bytes+buffer->dataStart, buffer->dataCount);
+ if (decoded < buffer->dataCount) {
+ // Adjust buffer for used bytes and then "unread them"
+ buffer->dataStart += decoded;
+ buffer->dataCount -= decoded;
+ aio->unread(buffer);
+ } else {
+ // Give whole buffer back to aio subsystem
+ aio->queueReadBuffer(buffer);
+ }
+}
+
+void SslTransport::write(SslIO&)
+{
+ if (context.getCodec().canEncode()) {
+ SslIO::BufferBase* buffer = aio->getQueuedBuffer();
+ if (buffer) {
+ size_t encoded = context.getCodec().encode(buffer->bytes, buffer->byteCount);
+
+ buffer->dataStart = 0;
+ buffer->dataCount = encoded;
+ aio->queueWrite(buffer);
+ }
+ }
+
+}
+
+void SslTransport::close()
+{
+ QPID_LOG(debug, id << " SslTransport closing...");
+ if (aio)
+ aio->queueWriteClose();
+}
+
+void SslTransport::eof(SslIO&)
+{
+ close();
+}
+
+void SslTransport::disconnected(SslIO&)
+{
+ close();
+ socketClosed(*aio, socket);
+}
+
+void SslTransport::socketClosed(SslIO&, const SslSocket&)
+{
+ if (aio)
+ aio->queueForDeletion();
+ context.closed();
+ QPID_LOG(debug, id << " Socket closed");
+}
+
+void SslTransport::abort()
+{
+ if (aio) {
+ // Established connection
+ aio->requestCallback(boost::bind(&SslTransport::eof, this, _1));
+ }
+}
+
+void SslTransport::activateOutput()
+{
+ if (aio) aio->notifyPendingWrite();
+}
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/SslTransport.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,74 @@
+#ifndef QPID_MESSAGING_AMQP_SSLTRANSPORT_H
+#define QPID_MESSAGING_AMQP_SSLTRANSPORT_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/amqp/Transport.h"
+#include "qpid/sys/Mutex.h"
+#include "qpid/sys/ssl/SslSocket.h"
+#include <boost/shared_ptr.hpp>
+
+namespace qpid {
+namespace sys {
+class ConnectionCodec;
+class Poller;
+namespace ssl {
+class SslIO;
+class SslIOBufferBase;
+}
+}
+
+namespace messaging {
+namespace amqp {
+class TransportContext;
+
+class SslTransport : public Transport
+{
+ public:
+ SslTransport(TransportContext&, boost::shared_ptr<qpid::sys::Poller> p);
+
+ void connect(const std::string& host, const std::string& port);
+
+ void activateOutput();
+ void abort();
+ void close();
+ void giveReadCredit(int32_t) {}
+ private:
+ qpid::sys::ssl::SslSocket socket;
+ TransportContext& context;
+ qpid::sys::ssl::SslIO* aio;
+ boost::shared_ptr<qpid::sys::Poller> poller;
+ bool closed;
+ std::string id;
+
+ void connected(const qpid::sys::ssl::SslSocket&);
+ void failed(const std::string& msg);
+ void read(qpid::sys::ssl::SslIO&, qpid::sys::ssl::SslIOBufferBase*);
+ void write(qpid::sys::ssl::SslIO&);
+ void eof(qpid::sys::ssl::SslIO&);
+ void disconnected(qpid::sys::ssl::SslIO&);
+ void socketClosed(qpid::sys::ssl::SslIO&, const qpid::sys::ssl::SslSocket&);
+
+ friend class DriverImpl;
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_SSLTRANSPORT_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.cpp Fri Oct 19 17:16:07 2012
@@ -0,0 +1,162 @@
+/*
+ *
+ * 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 "TcpTransport.h"
+#include "ConnectionContext.h"
+#include "qpid/sys/AsynchIO.h"
+#include "qpid/sys/ConnectionCodec.h"
+#include "qpid/sys/Poller.h"
+#include "qpid/log/Statement.h"
+#include <boost/bind.hpp>
+#include <boost/format.hpp>
+
+using namespace qpid::sys;
+
+namespace qpid {
+namespace messaging {
+namespace amqp {
+// Static constructor which registers connector here
+namespace {
+Transport* create(TransportContext& c, Poller::shared_ptr p)
+{
+ return new TcpTransport(c, p);
+}
+
+struct StaticInit
+{
+ StaticInit()
+ {
+ Transport::add("tcp", &create);
+ };
+} init;
+}
+
+TcpTransport::TcpTransport(TransportContext& c, boost::shared_ptr<Poller> p) : context(c), connector(0), aio(0), poller(p) {}
+
+void TcpTransport::connect(const std::string& host, const std::string& port)
+{
+ assert(!connector);
+ assert(!aio);
+ connector = AsynchConnector::create(
+ socket,
+ host, port,
+ boost::bind(&TcpTransport::connected, this, _1),
+ boost::bind(&TcpTransport::failed, this, _3));
+
+ connector->start(poller);
+}
+
+void TcpTransport::failed(const std::string& msg)
+{
+ QPID_LOG(debug, "Failed to connect: " << msg);
+ connector = 0;
+ socket.close();
+ context.closed();
+}
+
+void TcpTransport::connected(const Socket&)
+{
+ context.opened();
+ connector = 0;
+ aio = AsynchIO::create(socket,
+ boost::bind(&TcpTransport::read, this, _1, _2),
+ boost::bind(&TcpTransport::eof, this, _1),
+ boost::bind(&TcpTransport::disconnected, this, _1),
+ boost::bind(&TcpTransport::socketClosed, this, _1, _2),
+ 0, // nobuffs
+ boost::bind(&TcpTransport::write, this, _1));
+ aio->createBuffers(std::numeric_limits<uint16_t>::max());//note: AMQP 1.0 _can_ handle large frame sizes
+ id = boost::str(boost::format("[%1%]") % socket.getFullAddress());
+ aio->start(poller);
+}
+
+void TcpTransport::read(AsynchIO&, AsynchIO::BufferBase* buffer)
+{
+ int32_t decoded = context.getCodec().decode(buffer->bytes+buffer->dataStart, buffer->dataCount);
+ if (decoded < buffer->dataCount) {
+ // Adjust buffer for used bytes and then "unread them"
+ buffer->dataStart += decoded;
+ buffer->dataCount -= decoded;
+ aio->unread(buffer);
+ } else {
+ // Give whole buffer back to aio subsystem
+ aio->queueReadBuffer(buffer);
+ }
+}
+
+void TcpTransport::write(AsynchIO&)
+{
+ if (context.getCodec().canEncode()) {
+ AsynchIO::BufferBase* buffer = aio->getQueuedBuffer();
+ if (buffer) {
+ size_t encoded = context.getCodec().encode(buffer->bytes, buffer->byteCount);
+
+ buffer->dataStart = 0;
+ buffer->dataCount = encoded;
+ aio->queueWrite(buffer);
+ }
+ }
+
+}
+
+void TcpTransport::close()
+{
+ QPID_LOG(debug, id << " TcpTransport closing...");
+ if (aio)
+ aio->queueWriteClose();
+}
+
+void TcpTransport::eof(AsynchIO&)
+{
+ close();
+}
+
+void TcpTransport::disconnected(AsynchIO&)
+{
+ close();
+ socketClosed(*aio, socket);
+}
+
+void TcpTransport::socketClosed(AsynchIO&, const Socket&)
+{
+ if (aio)
+ aio->queueForDeletion();
+ context.closed();
+ QPID_LOG(debug, id << " Socket closed");
+}
+
+void TcpTransport::abort()
+{
+ if (aio) {
+ // Established connection
+ aio->requestCallback(boost::bind(&TcpTransport::eof, this, _1));
+ } else if (connector) {
+ // We're still connecting
+ connector->stop();
+ failed("Connection timedout");
+ }
+}
+
+void TcpTransport::activateOutput()
+{
+ if (aio) aio->notifyPendingWrite();
+}
+
+}}} // namespace qpid::messaging::amqp
Added: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h?rev=1400178&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TcpTransport.h Fri Oct 19 17:16:07 2012
@@ -0,0 +1,71 @@
+#ifndef QPID_MESSAGING_AMQP_TCPTRANSPORT_H
+#define QPID_MESSAGING_AMQP_TCPTRANSPORT_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/amqp/Transport.h"
+#include "qpid/sys/Mutex.h"
+#include "qpid/sys/Socket.h"
+#include <boost/shared_ptr.hpp>
+
+namespace qpid {
+namespace sys {
+class ConnectionCodec;
+class AsynchConnector;
+class AsynchIO;
+class AsynchIOBufferBase;
+class Poller;
+}
+namespace messaging {
+namespace amqp {
+class TransportContext;
+
+class TcpTransport : public Transport
+{
+ public:
+ TcpTransport(TransportContext&, boost::shared_ptr<qpid::sys::Poller>);
+
+ void connect(const std::string& host, const std::string& port);
+
+ void activateOutput();
+ void abort();
+ void close();
+ void giveReadCredit(int32_t) {}
+
+ private:
+ qpid::sys::Socket socket;
+ TransportContext& context;
+ qpid::sys::AsynchConnector* connector;
+ qpid::sys::AsynchIO* aio;
+ boost::shared_ptr<qpid::sys::Poller> poller;
+ std::string id;
+
+ void connected(const qpid::sys::Socket&);
+ void failed(const std::string& msg);
+ void read(qpid::sys::AsynchIO&, qpid::sys::AsynchIOBufferBase*);
+ void write(qpid::sys::AsynchIO&);
+ void eof(qpid::sys::AsynchIO&);
+ void disconnected(qpid::sys::AsynchIO&);
+ void socketClosed(qpid::sys::AsynchIO&, const qpid::sys::Socket&);
+};
+}}} // namespace qpid::messaging::amqp
+
+#endif /*!QPID_MESSAGING_AMQP_TCPTRANSPORT_H*/
Copied: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.cpp (from r1400177, qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.cpp?p2=qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.cpp&p1=qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h&r1=1400177&r2=1400178&rev=1400178&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.cpp Fri Oct 19 17:16:07 2012
@@ -1,6 +1,3 @@
-#ifndef QPID_MESSAGING_SENDERIMPL_H
-#define QPID_MESSAGING_SENDERIMPL_H
-
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
@@ -10,9 +7,9 @@
* 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
@@ -21,27 +18,32 @@
* under the License.
*
*/
-#include "qpid/RefCounted.h"
+#include "qpid/messaging/amqp/Transport.h"
+#include "qpid/messaging/amqp/TransportContext.h"
+#include <map>
namespace qpid {
namespace messaging {
+namespace amqp {
+namespace {
+typedef std::map<std::string, Transport::Factory*> Registry;
-class Message;
-class Session;
+Registry& theRegistry()
+{
+ static Registry factories;
+ return factories;
+}
+}
-class SenderImpl : public virtual qpid::RefCounted
+Transport* Transport::create(const std::string& name, TransportContext& context, boost::shared_ptr<qpid::sys::Poller> poller)
+{
+ Registry::const_iterator i = theRegistry().find(name);
+ if (i != theRegistry().end()) return (i->second)(context, poller);
+ else return 0;
+}
+void Transport::add(const std::string& name, Factory* factory)
{
- public:
- virtual ~SenderImpl() {}
- virtual void send(const Message& message, bool sync) = 0;
- virtual void close() = 0;
- virtual void setCapacity(uint32_t) = 0;
- virtual uint32_t getCapacity() = 0;
- virtual uint32_t getUnsettled() = 0;
- virtual const std::string& getName() const = 0;
- virtual Session getSession() const = 0;
- private:
-};
-}} // namespace qpid::messaging
+ theRegistry()[name] = factory;
+}
-#endif /*!QPID_MESSAGING_SENDERIMPL_H*/
+}}} // namespace qpid::messaging::amqp
Copied: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.h (from r1400177, qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.h?p2=qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.h&p1=qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h&r1=1400177&r2=1400178&rev=1400178&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/Transport.h Fri Oct 19 17:16:07 2012
@@ -1,5 +1,5 @@
-#ifndef QPID_MESSAGING_SENDERIMPL_H
-#define QPID_MESSAGING_SENDERIMPL_H
+#ifndef QPID_MESSAGING_AMQP_TRANSPORT_H
+#define QPID_MESSAGING_AMQP_TRANSPORT_H
/*
*
@@ -10,9 +10,9 @@
* 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
@@ -21,27 +21,28 @@
* under the License.
*
*/
-#include "qpid/RefCounted.h"
+#include "qpid/sys/OutputControl.h"
+#include <boost/shared_ptr.hpp>
namespace qpid {
+namespace sys {
+class Poller;
+}
namespace messaging {
+namespace amqp {
+class TransportContext;
-class Message;
-class Session;
-
-class SenderImpl : public virtual qpid::RefCounted
+class Transport : public qpid::sys::OutputControl
{
public:
- virtual ~SenderImpl() {}
- virtual void send(const Message& message, bool sync) = 0;
+ virtual ~Transport() {}
+ virtual void connect(const std::string& host, const std::string& port) = 0;
virtual void close() = 0;
- virtual void setCapacity(uint32_t) = 0;
- virtual uint32_t getCapacity() = 0;
- virtual uint32_t getUnsettled() = 0;
- virtual const std::string& getName() const = 0;
- virtual Session getSession() const = 0;
- private:
+
+ typedef Transport* Factory(TransportContext&, boost::shared_ptr<qpid::sys::Poller>);
+ static Transport* create(const std::string& name, TransportContext&, boost::shared_ptr<qpid::sys::Poller>);
+ static void add(const std::string& name, Factory* factory);
};
-}} // namespace qpid::messaging
+}}} // namespace qpid::messaging::amqp
-#endif /*!QPID_MESSAGING_SENDERIMPL_H*/
+#endif /*!QPID_MESSAGING_AMQP_TRANSPORT_H*/
Copied: qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TransportContext.h (from r1400177, qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TransportContext.h?p2=qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TransportContext.h&p1=qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h&r1=1400177&r2=1400178&rev=1400178&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/messaging/SenderImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/messaging/amqp/TransportContext.h Fri Oct 19 17:16:07 2012
@@ -1,5 +1,5 @@
-#ifndef QPID_MESSAGING_SENDERIMPL_H
-#define QPID_MESSAGING_SENDERIMPL_H
+#ifndef QPID_MESSAGING_AMQP_TRANSPORTCONTEXT_H
+#define QPID_MESSAGING_AMQP_TRANSPORTCONTEXT_H
/*
*
@@ -10,9 +10,9 @@
* 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
@@ -21,27 +21,27 @@
* under the License.
*
*/
-#include "qpid/RefCounted.h"
-
namespace qpid {
+namespace sys {
+class Codec;
+}
namespace messaging {
+namespace amqp {
-class Message;
-class Session;
-
-class SenderImpl : public virtual qpid::RefCounted
+/**
+ * Interface to be supplied by 'users' of Transport interface, in
+ * order to provide codec and handle callbaskc for opening and closing
+ * of connection.
+ */
+class TransportContext
{
public:
- virtual ~SenderImpl() {}
- virtual void send(const Message& message, bool sync) = 0;
- virtual void close() = 0;
- virtual void setCapacity(uint32_t) = 0;
- virtual uint32_t getCapacity() = 0;
- virtual uint32_t getUnsettled() = 0;
- virtual const std::string& getName() const = 0;
- virtual Session getSession() const = 0;
+ virtual ~TransportContext() {}
+ virtual qpid::sys::Codec& getCodec() = 0;
+ virtual void closed() = 0;
+ virtual void opened() = 0;
private:
};
-}} // namespace qpid::messaging
+}}} // namespace qpid::messaging::amqp
-#endif /*!QPID_MESSAGING_SENDERIMPL_H*/
+#endif /*!QPID_MESSAGING_AMQP_TRANSPORTCONTEXT_H*/
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org