// work_queue work items is are automatically dequeued and called by proton
 // This function is called because it was queued by send()
 void do_send(const proton::message& m) {
     sender_.send(m);
     std::lock_guard<std::mutex> l(lock_);
     --queued_;                    // work item was consumed from the work_queue
     credit_ = sender_.credit();   // update credit
     sender_ready_.notify_all();       // Notify senders we have space on queue
 }

 void on_sender_open(proton::sender &sender) override {
     if (sender.source().dynamic()) {
         std::string addr = generate_address();
         sender.open(proton::sender_options().source(proton::source_options().address(addr)));
         senders[addr] = sender;
     }
 }

 void on_sender_open(proton::sender& s) override {
     auto a = s.source().address();
     std::cout << s << ": New incoming sender: source address: " << a << "\n";
     if (a==c_quit) {
         s.container().stop();
     }
 }

 void send_available_messages(proton::sender &s) {
     for (int i = sequence; available && s.credit() > 0; i++) {
         std::ostringstream mbody;
         mbody << "flow_sender message " << sequence++;
         proton::message m(mbody.str());
         s.send(m);
         available--;
     }
 }

 void on_sendable(proton::sender &sender) override {
     while (sender.credit() && sent_ < total_) {
         id_value_ = sent_ + 1;
         message_.correlation_id(id_value_);
         message_.creation_time(proton::timestamp::now());
         sender.send(message_);
         sent_++;
     }
 }

 void on_sendable(proton::event &e, proton::sender &sender) override {
     while (sender.credit() && sent < total) {
         proton::message msg;
         msg.id(sent + 1);
         std::map<std::string, int> m;
         m["sequence"] = sent+1;
         msg.body(m);
         sender.send(msg);
         sent++;
     }
 }

    void send_request() {
        proton::message req;
        req.body(requests.front());
        req.reply_to(receiver.remote_source().address());

        sender.send(req);
    }

 void unsubscribe(proton::sender lnk) {
     std::string address = lnk.local_source().address();
     
     if (queues_.get(address).unsubscribe(lnk)) {
         queues_.erase(address);
     }
 }

 void unsubscribe (proton::sender &lnk) {
     std::string address = lnk.source().address();
     queue_map::iterator it = queues.find(address);
     if (it != queues.end() && it->second->unsubscribe(lnk)) {
         delete it->second;
         queues.erase(it);
     }
 }

 // Only called if we have credit. Return true if we sent a message.
 bool do_send(queue* q, proton::sender &s) {
     proton::message m;
     bool popped =  q->pop(m, has_messages_callback_);
     if (popped)
         s.send(m);
     /// if !popped the queue has saved the callback for later.
     return popped;
 }

 void on_sender_open(proton::sender &sender) override {
     if (sender.remote_source().dynamic()) {
         sender.local_source().address(generate_address());
         senders[sender.local_source().address()] = sender;
     }
 }

 void on_sendable(proton::event &e, proton::sender &s) override {
     proton::message m;
     m.body("Hello World!");
     s.send(m);
     s.close();
 }

 void on_sendable(proton::sender& s) override {
     std::lock_guard<std::mutex> l(lock_);
     credit_ = s.credit();
     sender_ready_.notify_all(); // Notify senders we have credit
 }

 void on_sender_open(proton::sender& s) override {
     // Make sure sender_ and work_queue_ are set atomically
     std::lock_guard<std::mutex> l(lock_);
     sender_ = s;
     work_queue_ = &s.work_queue();
 }

 // Thread safe
 void close() {
     work_queue()->add([=]() { sender_.connection().close(); });
 }

 // A sender sends messages from a queue to a subscriber.
 void on_sender_open(proton::sender &sender) override {
     queue *q = sender.source().dynamic() ?
         queues_.dynamic() : queues_.get(sender.source().address());
     std::cout << "sending from " << q->name() << std::endl;
 }

 void on_start(proton::event &e) {
     sender = e.container().open_sender(url);
     // Create a receiver with a dynamically chosen unique address.
     receiver = sender.connection().open_receiver("", proton::link_options().dynamic_address(true));
 }

 void subscribe(proton::sender &c) {
     consumers.push_back(c.ptr());
 }