qpid-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Darryl L. Pierce" <dpie...@redhat.com>
Subject [PATCH 5/5] Ruby example for setting up a mail exchange server.
Date Fri, 06 Jul 2012 17:47:52 GMT
From: "Darryl L. Pierce" <dpierce@redhat.com>

 * server.rb - Acts as the intermediary, storing messages in mailboxes.
 * post.rb   - Submits messages for storing in mailboxes.
 * fetch.rb  - Retrieves messages from mailboxes.
---
 examples/ruby/EXAMPLES      |    4 +
 examples/ruby/fetch.rb      |  204 +++++++++++++++++++++++++++
 examples/ruby/mailserver.rb |  322 +++++++++++++++++++++++++++++++++++++++++++
 examples/ruby/post.rb       |  191 +++++++++++++++++++++++++
 4 files changed, 721 insertions(+)
 create mode 100644 examples/ruby/fetch.rb
 create mode 100644 examples/ruby/mailserver.rb
 create mode 100644 examples/ruby/post.rb

diff --git a/examples/ruby/EXAMPLES b/examples/ruby/EXAMPLES
index f2eaa61..d88fb3d 100644
--- a/examples/ruby/EXAMPLES
+++ b/examples/ruby/EXAMPLES
@@ -4,3 +4,7 @@ EXAMPLES:
    Demonstrates the use of the messenger APIs for sending and receiving
    messages from point to point.
 
+ * server.rb, post.rb, fetch.rb:
+   Demonstrates setting up a mail exchange via the server app and then posting
+   and fetching messages to specific mailboxes via post and fetch,
+   respectively.
diff --git a/examples/ruby/fetch.rb b/examples/ruby/fetch.rb
new file mode 100644
index 0000000..6bff13f
--- /dev/null
+++ b/examples/ruby/fetch.rb
@@ -0,0 +1,204 @@
+#!/usr/bin/env ruby
+#
+# 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.
+
+require 'cproton'
+require 'optparse'
+
+$options = {
+  :hostname => "0.0.0.0",
+  :port => "5672",
+  :count => 1,
+  :verbose => false
+}
+messages = []
+
+OptionParser.new do |opts|
+  opts.banner = "Usage: #{File.basename $0} [options] <mailbox>"
+
+  opts.on("-s", "--server <address>", String, :REQUIRED,
+          "Address of the server") do |server|
+  end
+
+  opts.on("-p", "--port <port>", Integer, :REQUIRED,
+          "Port on the server") do |port|
+    $options[:port] = port
+  end
+
+  opts.on("-c", "--count <#>", Integer, :REQUIRED,
+          "Number of messages to read from the mailbox") do |count|
+    $options[:count] = count
+  end
+
+  opts.on("-v", "--verbose", :NONE,
+          "Turn on extra trace messages.") do |verbose|
+    $options[:verbose] = true
+  end
+
+  begin
+    ARGV << "-h" if ARGV.empty?
+    opts.parse!
+  rescue OptionParser::ParseError => error
+    STDERR.puts error.message, "\n", opts
+    exit 1
+  end
+
+  $options[:mailbox] = ARGV.first unless ARGV.empty?
+
+  abort("No mailbox specified.") if $options[:mailbox].nil?
+
+end
+
+def log(text, return_code = 0)
+  STDOUT.puts "#{Time.new}: #{text}" if $options[:verbose] || return_code.nonzero?
+
+  # if we were given a non-zero code, then exit
+  exit return_code unless return_code.zero?
+end
+
+class FetchClient
+  attr_reader :sasl
+  attr_reader :link
+  attr_reader :conn
+
+  def initialize(hostname, port, mailbox)
+    @hostname = hostname
+    @port = port
+    @mailbox = mailbox
+  end
+
+  def setup
+    # setup a driver connection to the server
+    log "Connecting to server host = #{@hostname}:#{@port}"
+    @driver = Cproton::pn_driver
+    @cxtr = Cproton::pn_connector(@driver, @hostname, @port, nil)
+
+    # configure SASL
+    @sasl = Cproton::pn_connector_sasl @cxtr
+    Cproton::pn_sasl_mechanisms @sasl, "ANONYMOUS"
+    Cproton::pn_sasl_client @sasl
+
+    # inform the engine about the connection, and link the driver to it.
+    @conn = Cproton::pn_connection
+    Cproton::pn_connector_set_connection @cxtr, @conn
+
+    # create a session and link for receiving from the mailbox
+    log "Fetching from the mailbox = #{@mailbox}"
+    @ssn = Cproton::pn_session @conn
+    @link = Cproton::pn_receiver @ssn, "receiver"
+    Cproton::pn_set_source @link, @mailbox
+
+    # now open all the engine endpoints
+    Cproton::pn_connection_open @conn
+    Cproton::pn_session_open @ssn
+    Cproton::pn_link_open @link
+  end
+
+  def wait
+    log "Waiting for events..."
+    Cproton::pn_connector_process @cxtr
+    Cproton::pn_driver_wait @driver, -1
+    Cproton::pn_connector_process @cxtr
+    log "...waiting done!"
+  end
+
+  def settle
+    # locally settle any remotely settled deliveries
+    d = Cproton::pn_unsettled_head @link
+    while d && !Cproton::pn_readable(d)
+      # delivery that has not yet been read
+      _next = Cproton::pn_unsettled_next d
+      Cproton::pn_settle(d) if Cproton::pn_remote_settled(d)
+      d = _next
+    end
+  end
+
+end
+
+if __FILE__ == $PROGRAM_NAME
+  receiver = FetchClient.new($options[:hostname],
+                             $options[:port],
+                             $options[:mailbox])
+
+  receiver.setup
+
+  # wait until we authenticate with the server
+  while ![Cproton::PN_SASL_PASS, Cproton::PN_SASL_FAIL].include? Cproton::pn_sasl_state(receiver.sasl)
+    receiver.wait
+  end
+
+  if Cproton::pn_sasl_state(receiver.sasl) == Cproton::PN_SASL_FAIL
+    log "ERROR: Authentication failure", -1
+  end
+
+  # wait until the server has opened the connection
+  while ((Cproton::pn_link_state(receiver.link) & Cproton::PN_REMOTE_ACTIVE) != Cproton::PN_REMOTE_ACTIVE)
+    receiver.wait
+  end
+
+  # check if the server recognizes the mailbox, fail if it does not
+  if Cproton::pn_remote_source(receiver.link) != $options[:mailbox]
+    log "ERROR: mailbox #{$options[:mailbox]} does not exist!", -2
+  end
+
+  # Allow the server to send the expected number of messages to the receiver
+  # by setting the credit to the expected count
+  Cproton::pn_flow(receiver.link, $options[:count])
+
+  # main loop: continue fetching messages until all the expected number of
+  # messages have been retrieved
+
+  while Cproton::pn_credit(receiver.link) > 0
+    # wait for some messages to arrive
+    receiver.wait if Cproton::pn_queued(receiver.link).zero?
+
+    # read all queued deliveries
+    while Cproton::pn_queued(receiver.link) > 0
+      delivery = Cproton::pn_current(receiver.link)
+
+      # read all bytes of message
+      (rc, msg) = Cproton::pn_recv(receiver.link, Cproton::pn_pending(delivery))
+      log "Received count/status=#{rc}"
+
+      log("ERROR: Receive failed (#{rc}), exiting...", -3) if rc < 0
+
+      puts "#{msg}"
+
+      # let the server know we accept the message
+      Cproton::pn_disposition(delivery, Cproton::PN_ACCEPTED)
+
+      # go to the next deliverable
+      Cproton::pn_advance(receiver.link)
+    end
+
+    receiver.settle
+  end
+
+  # block until any leftover deliveries are settled
+  while Cproton::pn_unsettled(receiver.link) > 0
+    receiver.wait
+    receiver.settle
+  end
+
+  # we're done,c lose and wait for the remote to close also
+  Cproton::pn_connection_close(receiver.conn)
+  while ((Cproton::pn_connection_state(receiver.conn) & Cproton::PN_REMOTE_CLOSED) !=
Cproton::PN_REMOTE_CLOSED)
+    receiver.wait
+  end
+
+end
diff --git a/examples/ruby/mailserver.rb b/examples/ruby/mailserver.rb
new file mode 100644
index 0000000..582d164
--- /dev/null
+++ b/examples/ruby/mailserver.rb
@@ -0,0 +1,322 @@
+#!/usr/bin/env ruby
+#
+# 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.
+#
+
+require 'cproton'
+require 'optparse'
+
+FAILED         = 0
+CONNECTION_UP  = 1
+AUTHENTICATING = 2
+
+$options  = {
+  :verbose => false,
+  :hostname => "0.0.0.0",
+  :port => "5672"
+}
+
+OptionParser.new do |opts|
+  opts.banner = "Usage: mailserver [options] <server-address> [<server-port>]"
+
+  opts.on("-v", "--verbose", :NONE,
+          "Print status messages to stdout") do |f|
+    $options[:verbose] = true
+  end
+
+  opts.parse!
+
+  $options[:hostname] = ARGV[0] if ARGV.length > 0
+  $options[:port] = ARGV[1] if ARGV.length > 1
+end
+
+def log(text)
+  STDOUT.puts "#{Time.new}: #{text}" if $options[:verbose]
+end
+
+class MailServer
+  def initialize(hostname, port, verbose)
+    @hostname = hostname
+    @port = port
+    @verbose = verbose
+    @counter = 0
+    @mailboxes = {}
+
+    @driver = nil
+  end
+
+  def setup
+    @driver = Cproton::pn_driver
+    @listener = Cproton::pn_listener @driver, @hostname, @port, nil
+    raise "Error: could not listen on #{@hostname}:#{@port}" if @listener.nil?
+  end
+
+  def wait
+    log "Driver sleep."
+    Cproton::pn_driver_wait @driver, -1
+    log "Driver wakeup."
+  end
+
+  def accept_connection_requests
+    l = Cproton::pn_driver_listener @driver
+
+    while !l.nil?
+      log "Accepting connection."
+      cxtr = Cproton::pn_listener_accept l
+      Cproton::pn_connector_set_context cxtr, AUTHENTICATING
+      l = Cproton::pn_driver_listener @driver
+    end
+  end
+
+  def process_connections
+    cxtr = Cproton::pn_driver_connector @driver
+
+    while cxtr
+      log "Process connector"
+
+      Cproton::pn_connector_process cxtr
+
+      state = Cproton::pn_connector_context cxtr
+      case state
+      when AUTHENTICATING
+        log "Authenticating..."
+        self.authenticate_connector cxtr
+
+      when CONNECTION_UP
+        log "Connection established..."
+        self.service_connector cxtr
+
+      else
+        raise "Unknown connection state #{state}"
+      end
+
+      Cproton::pn_connector_process cxtr
+
+      if Cproton::pn_connector_closed cxtr
+        log "Closing connector."
+        Cproton::pn_connector_free cxtr
+      end
+
+      cxtr = Cproton::pn_driver_connector @driver
+
+    end
+  end
+
+  def authenticate_connector(cxtr)
+    log "Authenticating..."
+
+    sasl = Cproton::pn_connector_sasl cxtr
+    state = Cproton::pn_sasl_state sasl
+    while [Cproton::PN_SASL_CONF, Cproton::PN_SASL_STEP].include? state
+
+      case state
+      when Cproton::PN_SASL_CONF
+        log "Authenticating-CONF..."
+        Cproton::pn_sasl_mechanisms sasl, "ANONYMOUS"
+        Cproton::pn_sasl_server sasl
+
+      when Cproton::PN_SASL_STEP
+        log "Authenticating-STEP..."
+        mech = Cproton::pn_sasl_remote_mechanisms sasl
+        if mech == "ANONYMOUS"
+          Cproton::pn_sasl_done sasl, Cproton::PN_SASL_OK
+        else
+          Cproton::pn_sasl_done sasl, Cproton::PN_SASL_AUTH
+        end
+      end
+
+      state = Cproton::pn_sasl_state sasl
+
+    end
+
+    case state
+    when Cproton::PN_SASL_PASS
+      Cproton::pn_connector_set_connection cxtr, Cproton::pn_connection
+      Cproton::pn_connector_set_context cxtr, CONNECTION_UP
+      log "Authentication-PASSED"
+
+    when Cproton::PN_SASL_FAIL
+      Cproton::pn_connector_set_context cxtr, FAILED
+      log "Authentication-FAILED"
+
+    else
+      log "Authentication-PENDING"
+    end
+
+  end
+
+  def service_connector(cxtr)
+    log "I/O processing starting."
+
+    conn = Cproton::pn_connector_connection cxtr
+    if ((Cproton::pn_connection_state(conn) & Cproton::PN_LOCAL_UNINIT) == Cproton::PN_LOCAL_UNINIT)
+      log "Connection opened."
+      Cproton::pn_connection_open(conn)
+    end
+
+    ssn = Cproton::pn_session_head conn, Cproton::PN_LOCAL_UNINIT
+    while ssn
+      Cproton::pn_session_open ssn
+      log "Session opened."
+      ssn = Cproton::pn_session_next ssn, Cproton::PN_LOCAL_UNINIT
+    end
+
+    link = Cproton::pn_link_head conn, Cproton::PN_LOCAL_UNINIT
+    while link
+      setup_link link
+      link = Cproton::pn_link_next link, Cproton::PN_LOCAL_UNINIT
+    end
+
+    delivery = Cproton::pn_work_head conn
+    while delivery
+      log "Process delivery #{Cproton::pn_delivery_tag delivery}"
+      if Cproton::pn_readable delivery
+        process_receive delivery
+      elsif Cproton::pn_writable delivery
+        send_message delivery
+      end
+
+      if Cproton::pn_updated delivery
+        log "Remote disposition for #{Cproton::pn_delivery_tag delivery}: #{Cproton::pn_remote_disposition
delivery}"
+        Cproton::pn_settle(delivery) if Cproton::pn_remote_disposition delivery
+      end
+
+      delivery = Cproton::pn_work_next delivery
+    end
+
+    link = Cproton::pn_link_head conn, Cproton::PN_LOCAL_ACTIVE | Cproton::PN_REMOTE_CLOSED
+    while link
+      Cproton::pn_link_close link
+      log "Link closed."
+      link = Cproton::pn_link_next link, Cproton::PN_LOCAL_ACTIVE | Cproton::PN_REMOTE_CLOSED
+    end
+
+    ssn = Cproton::pn_session_head conn, Cproton::PN_LOCAL_ACTIVE | Cproton::PN_REMOTE_CLOSED
+    while ssn
+      Cproton::pn_session_close ssn
+      log "Session closed."
+      ssn = Cproton::pn_session_next ssn, Cproton::PN_LOCAL_ACTIVE | Cproton::PN_REMOTE_CLOSED
+    end
+
+    if Cproton::pn_connection_state(conn) == (Cproton::PN_LOCAL_ACTIVE | Cproton::PN_REMOTE_CLOSED)
+      log "Connection closed."
+      Cproton::pn_connection_close conn
+    end
+
+  end
+
+  def process_receive(delivery)
+    link = Cproton::pn_link delivery
+    mbox = Cproton::pn_remote_target link
+    (rc, msg) = Cproton::pn_recv link, 1024
+
+    log "Message received #{rc}"
+
+    while rc >= 0
+      unless @mailboxes.include? mbox
+        log "Error: cannot send to mailbox #{mbox} - dropping message."
+      else
+        @mailboxes[mbox] << msg
+        log "Mailbox #{mbox} contains: #{@mailboxes[mbox].size}"
+      end
+
+      (rc, msg) = Cproton::pn_recv link, 1024
+    end
+
+    log "Messages accepted."
+
+    Cproton::pn_disposition delivery, Cproton::PN_ACCEPTED
+    Cproton::pn_settle delivery
+    Cproton::pn_advance link
+
+    Cproton::pn_flow(link, 1) if Cproton::pn_credit(link).zero?
+  end
+
+  def send_message(delivery)
+    link = Cproton::pn_link delivery
+    mbox = Cproton::pn_remote_source link
+    log "Request for Mailbox=#{mbox}"
+
+    if @mailboxes.include?(mbox) && !@mailboxes[mbox].empty?
+      msg = @mailboxes[mbox].first
+      @mailboxes[mbox].delete_at 0
+      log "Fetching message #{msg}"
+    else
+      log "Warning: mailbox #{mbox} is empty, sending empty message"
+      msg = ""
+    end
+
+    sent = Cproton::pn_send link, msg
+    log "Message sent: #{sent}"
+
+    if Cproton::pn_advance link
+      Cproton::pn_delivery link, "server-delivery-#{@counter}"
+      @counter += 1
+    end
+  end
+
+  def setup_link(link)
+    r_tgt = Cproton::pn_remote_target link
+    r_src = Cproton::pn_remote_source link
+
+    if Cproton::pn_is_sender link
+      log "Opening link to read from mailbox: #{r_src}"
+
+      unless @mailboxes.include? r_src
+        log "Error: mailbox #{r_src} does not exist!"
+        r_src = nil
+      end
+    else
+      log "Opening link to write to mailbox: #{r_tgt}"
+
+      @mailboxes[r_tgt] = [] unless @mailboxes.include? r_tgt
+    end
+
+    Cproton::pn_set_target link, r_tgt
+    Cproton::pn_set_source link, r_src
+
+    if Cproton::pn_is_sender link
+      Cproton::pn_delivery link, "server-delivery-#{@counter}"
+      @counter += 1
+    else
+      Cproton::pn_flow link, 1
+    end
+
+    Cproton::pn_link_open link
+
+  end
+
+end
+
+#------------------
+# Begin entry point
+#------------------
+
+if __FILE__ == $PROGRAM_NAME
+  server = MailServer.new($options[:hostname],
+                          $options[:port],
+                          $options[:verbose])
+
+  server.setup
+  loop do
+    server.wait
+    server.accept_connection_requests
+    server.process_connections
+  end
+end
+
diff --git a/examples/ruby/post.rb b/examples/ruby/post.rb
new file mode 100644
index 0000000..dfff488
--- /dev/null
+++ b/examples/ruby/post.rb
@@ -0,0 +1,191 @@
+#!/usr/bin/env ruby
+#
+# 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.
+#
+
+require 'cproton'
+require 'optparse'
+
+
+$options  = {
+  :verbose => false,
+  :hostname => "0.0.0.0",
+  :port => "5672"
+}
+
+
+OptionParser.new do |opts|
+  opts.banner = "Usage: mailserver [options] <server-address> [<server-port>]
<message-string> [<message-string> ...]"
+
+  opts.on("-s", "--server", String, :REQUIRED,
+          "The server hostname (def. #{$options[:hostname]})") do |hostname|
+    $options[:hostname] = hostname
+  end
+
+  opts.on("-p", "--port", String, :REQUIRED,
+          "The server port (def. #{$options[:port]})") do |port|
+    $options[:post] = port
+  end
+
+  opts.on("-m", "--mailbox", String, :REQUIRED,
+          "Name of the mailbox on the server") do |mailbox|
+    $options[:mailbox] = mailbox
+  end
+
+  opts.on("-v", "--verbose", :NONE,
+             "Enable verbose output (def. #{$options[:verbose]})") do
+    $options[:verbose] = true
+  end
+
+  begin
+    ARGV << "-h" if ARGV.empty?
+    opts.parse!(ARGV)
+  rescue OptionParser::ParseError => error
+    STDERR.puts error.message, "\n", opts
+    exit 1
+  end
+
+  $options[:messages] = ARGV
+
+  abort "No mailbox specified." if $options[:mailbox].nil?
+  abort "No messages specified." if $options[:messages].empty?
+
+end
+
+def log(text)
+  printf "#{Time.new}: #{text}\n" if $options[:verbose]
+end
+
+
+class PostClient
+
+  attr_reader :sasl
+  attr_reader :link
+  attr_reader :conn
+
+  def initialize(hostname, port, mailbox, verbose)
+    @hostname = hostname
+    @port = port
+    @mailbox = mailbox
+    @verbose = verbose
+  end
+
+  def setup
+    log "Connection to server host = #{@hostname}:#{@port}"
+    @driver = Cproton::pn_driver
+    @cxtr = Cproton::pn_connector(@driver, @hostname, @port, nil)
+
+    # configure SASL
+    @sasl = Cproton::pn_connector_sasl @cxtr
+    Cproton::pn_sasl_mechanisms @sasl, "ANONYMOUS"
+    Cproton::pn_sasl_client @sasl
+
+    # inform the engine about the connection, and link the driver to it.
+    @conn = Cproton::pn_connection
+    Cproton::pn_connector_set_connection @cxtr, @conn
+
+    # create a session and link for receiving from the mailbox
+    log "Posting to mailbox = #{@mailbox}"
+    @ssn = Cproton::pn_session @conn
+    @link = Cproton::pn_sender @ssn, "sender"
+    Cproton::pn_set_target @link, @mailbox
+
+    # now open all the engien end points
+    Cproton::pn_connection_open @conn
+    Cproton::pn_session_open @ssn
+    Cproton::pn_link_open @link
+  end
+
+  def wait
+    log "Waiting for events..."
+    Cproton::pn_connector_process @cxtr
+    Cproton::pn_driver_wait @driver, -1
+    Cproton::pn_connector_process @cxtr
+    log "..waiting done!"
+  end
+
+  def settle
+    d = Cproton::pn_unsettled_head @link
+    while d
+      _next = Cproton::pn_unsettled_next d
+      disp = Cproton::pn_remote_disposition d
+
+      if disp.nonzero? && disp != Cproton::PN_ACCEPTED
+        log "Warning: message was not accepted by the remote!"
+      end
+
+      if disp || Cproton::pn_remote_settled(disp)
+        Cproton::pn_settle(d)
+      end
+
+      d = _next
+    end
+  end
+
+end
+
+
+if __FILE__ == $PROGRAM_NAME
+  sender = PostClient.new($options[:hostname],
+                          $options[:port],
+                          $options[:mailbox],
+                          $options[:verbose])
+  sender.setup
+
+  sender.wait while ![Cproton::PN_SASL_PASS, Cproton::PN_SASL_FAIL].include? Cproton::pn_sasl_state(sender.sasl)
+
+  abort "Error: Authentication failure" if Cproton::pn_sasl_state(sender.sasl) == Cproton::PN_SASL_FAIL
+
+  while !$options[:messages].empty? do
+
+    while Cproton::pn_credit(sender.link).zero?
+      log "Waiting for credit"
+      sender.wait
+    end
+
+    while Cproton::pn_credit(sender.link) > 0
+      msg = $options[:messages].first
+      $options[:messages].delete_at 0
+      log "Sending #{msg}"
+      d = Cproton::pn_delivery sender.link, "post-deliver-#{$options[:messages].length}"
+      rc = Cproton::pn_send(sender.link, msg)
+
+      abort "Error: sending message: #{msg}" if rc < 0
+
+      fail unless rc == msg.length
+
+      Cproton::pn_advance sender.link
+    end
+
+    sender.settle
+  end
+
+  while Cproton::pn_unsettled(sender.link) > 0
+    log "Settling things with the server..."
+    sender.wait
+    sender.settle
+    log "Finished settling."
+  end
+
+  Cproton::pn_connection_close sender.conn
+  while ((Cproton::pn_connection_state(sender.conn) & Cproton::PN_REMOTE_CLOSED) != Cproton::PN_REMOTE_CLOSED)
+    log "Waiting for things to become closed..."
+    sender.wait
+  end
+
+end
-- 
1.7.10.4


---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org


Mime
View raw message