Uploaded image for project: 'Qpid Proton'
  1. Qpid Proton
  2. PROTON-2000

[cpp] source timeout

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Closed
    • Priority: Major
    • Resolution: Invalid
    • Affects Version/s: None
    • Fix Version/s: None
    • Component/s: None
    • Labels:
      None

      Description

      Hi there,

      Receiver side (c++):

      #include <proton/messaging_handler.hpp>
      #include <proton/receiver.hpp>
      #include <proton/container.hpp>
      #include <proton/message.hpp>
      #include <proton/message_id.hpp>
      #include <proton/connection.hpp>
      #include <proton/delivery.hpp>
      
      
      #include <iostream>
      
      #include <unistd.h>
      #include <proton/target_options.hpp>
      
      
      //using namespace std;
      
      using namespace proton;
      /*
      *
      */
      class in : public proton::messaging_handler
      {
      private:
      receiver reiviever1;
      
      public:
      in() :
      messaging_handler(), reiviever1()
      {
      }
      void on_container_start(container& c) override
      {
      reiviever1 = c.open_receiver("amqp://127.0.0.1:5672/mbx");
      
      std::cout << "id of the container is " << c.id() << std::endl;
      
      }
      void on_message(delivery& d, message& m) override
      {
      /// Just print what you have for now.
      std::cout << "Received user <" << m.user() << ">" << std::endl;
      std::cout << "Received body <" << m.body() << ">" << std::endl;
      std::cout << "Received reply to <" << m.reply_to() << ">" << std::endl;
      std::cout << "Received ttl <" << m.ttl().milliseconds() << ">" << std::endl;
      std::cout << "Received subject <" << m.subject() << ">" << std::endl;
      if (!m.reply_to().empty())
      {
      std::cout << "sleeping for sometime before reply" << std::endl;
      sleep(5);
      
      proton::message r;
      r.to(m.reply_to());
      
      
      std::string text;
      text ="Reply form the receiver";
      r.body(text);
      
      r.correlation_id(m.correlation_id());
      
      proton::sender sr = d.connection().open_sender(m.reply_to());
      sr.send(r);
      sr.close();
      
      }
      std::cout << __func__ << ": Ended" << std::endl;
      
      }
      
      
      } ;
      
      
      int main(int argc, char** argv)
      {
      try
      {
      
      in i;
      proton::container(i).run();
      
      }
      catch (const std::exception & e )
      {
      std::cerr << e.what() << std::endl;
      }
      catch (...)
      {
      std::cout << "caught" << std::endl;
      }
      
      return 0;
      }
      
      
      

       

      Client side (sender): 

      #include <proton/messaging_handler.hpp>
      #include <proton/sender.hpp>
      #include <proton/receiver.hpp>
      #include <proton/container.hpp>
      #include <proton/message.hpp>
      #include <proton/sender_options.hpp>
      #include <proton/delivery.hpp>
      #include <iostream>
      #include <proton/source_options.hpp>
      #include <proton/receiver_options.hpp>
      #include <proton/target_options.hpp>
      #include <proton/message.hpp>
      
      //using namespace std;
      
      using namespace proton;
      /*
      *
      */
      class out : public proton::messaging_handler
      {
      private:
      sender _sender;
      receiver _receiver;
      
      public:
      out() :
      messaging_handler(), _sender(), _receiver()
      {
      }
      void on_container_start(container& c) override
      {
      /// Set the sender.
      /// open to send
      _sender = c.open_sender("amqp://127.0.0.1:5672/mbx");
      
      
      _receiver = _sender.connection().open_receiver("",
      receiver_options().source(
      source_options().timeout(duration(500)).dynamic(true)
      )
      );
      
      }
      void on_receiver_open(receiver& r) override
      {
      std::cout << __func__ << " called" << std::endl;
      send();
      
      }
      ///
      void send()
      {
      message msg;
      msg.user("Sender User");
      msg.subject("Sbj");
      std::string text;
      text = "Hello there!";
      msg.body(text);
      
      
      msg.reply_to(_receiver.source().address());
      
      std::cout << "out msg user <" << msg.user() << ">" << std::endl;
      std::cout << "out msg body <" << msg.body() << ">" << std::endl;
      std::cout << "out msg reply to <" << msg.reply_to() << ">" << std::endl;
      
      
      _sender.send(msg);
      _sender.close();
      
      /// do not close the connection yet.
      //s.connection().close();
      
      
      }
      void on_message(delivery& d, message& m) override
      {
      /// Just print what you have for now.
      std::cout << "RPL user <" << m.user() << ">" << std::endl;
      std::cout << "RPL body <" << m.body() << ">" << std::endl;
      std::cout << "RPL reply to <" << m.reply_to() << ">" << std::endl;
      
      d.connection().close();
      
      
      
      //
      int main(int argc, char** argv)
      {
      
      try
      {
      out o;
      proton::container(o).run();
      
      }
      catch (const std::exception & e )
      {
      std::cerr << e.what() << std::endl;
      }
      catch (...)
      {
      std::cout << "caught" << std::endl;
      }
      
      return 0;
      }
      
      
      

       

      What I'm trying to do is if the sender is not receiving an answer withing some time then timeout waiting.

      Please correct me if I'm wrong. I think this should done by source timeout option.

      This one doesn't seem to work

       

      Work environment :

      Ubuntu 18.04 

      libraries installed by apt 

      apt install libqpid-proton-cpp12-dev

      which is proton 0.22

       

       

      Thanks in advance,

      Panos

       

       

        Attachments

          Activity

            People

            • Assignee:
              Unassigned
              Reporter:
              sakellarioupk Panos Sakellariou
            • Votes:
              0 Vote for this issue
              Watchers:
              2 Start watching this issue

              Dates

              • Created:
                Updated:
                Resolved: