#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 #include #include 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 parent; const std::string name; const qpid::messaging::Address address; State state; std::auto_ptr sink; qpid::client::AsyncSession session; std::string destination; std::string routingKey; typedef boost::ptr_deque 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 void execute() { F f(*this); parent->execute(f); } template bool execute1(P p) { F f(*this, p); return parent->execute(f); } }; }}} // namespace qpid::client::amqp0_10 #endif /*!QPID_CLIENT_AMQP0_10_SENDERIMPL_H*/