You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2015/02/24 15:22:25 UTC
[1/3] qpid-proton git commit: Revert "PROTON-825: Reactor should
yield before quiescing."
Repository: qpid-proton
Updated Branches:
refs/heads/master bbc04608b -> dc27a7b1c
Revert "PROTON-825: Reactor should yield before quiescing."
This reverts commit 1455f32575921ac8ed90cce539273233058f2b70.
There is no need to put this logic in the reactor process loop, instead
an application that needs the reactor to yield the thread can call
pn_reactor_yield. New fix for SyncRequestResponse will follow.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/14bdfbe0
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/14bdfbe0
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/14bdfbe0
Branch: refs/heads/master
Commit: 14bdfbe045f036c99cb5d9fb3a6117fc1b1dd863
Parents: bbc0460
Author: Alan Conway <ac...@redhat.com>
Authored: Tue Feb 24 08:32:27 2015 -0500
Committer: Alan Conway <ac...@redhat.com>
Committed: Tue Feb 24 08:32:27 2015 -0500
----------------------------------------------------------------------
examples/python/server.py | 5 +-
examples/python/sync_client.py | 80 +++++++++++++++++++--------
proton-c/bindings/python/proton/utils.py | 2 +-
proton-c/include/proton/reactor.h | 9 ---
proton-c/src/reactor/reactor.c | 5 --
5 files changed, 60 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/14bdfbe0/examples/python/server.py
----------------------------------------------------------------------
diff --git a/examples/python/server.py b/examples/python/server.py
index a8d63f8..fc9ac8d 100755
--- a/examples/python/server.py
+++ b/examples/python/server.py
@@ -46,11 +46,10 @@ class Server(MessagingHandler):
if not sender:
sender = self.container.create_sender(self.conn, event.message.reply_to)
self.senders[event.message.reply_to] = sender
- sender.send(Message(address=event.message.reply_to, body=event.message.body.upper(),
- correlation_id=event.message.correlation_id))
+ sender.send(Message(address=event.message.reply_to, body=event.message.body.upper()))
try:
- Container(Server("0.0.0.0:5672", "examples")).run()
+ Container(Server("localhost:5672", "examples")).run()
except KeyboardInterrupt: pass
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/14bdfbe0/examples/python/sync_client.py
----------------------------------------------------------------------
diff --git a/examples/python/sync_client.py b/examples/python/sync_client.py
index 82cd85f..86cd7c2 100755
--- a/examples/python/sync_client.py
+++ b/examples/python/sync_client.py
@@ -24,31 +24,65 @@ Demonstrates the client side of the synchronous request-response pattern
"""
-import optparse
from proton import Message, Url, ConnectionException, Timeout
-from proton.utils import SyncRequestResponse, BlockingConnection
+from proton.utils import BlockingConnection
from proton.handlers import IncomingMessageHandler
import sys
-parser = optparse.OptionParser(usage="usage: %prog [options]",
- description="Send requests to the supplied address and print responses.")
-parser.add_option("-a", "--address", default="localhost:5672/examples",
- help="address to which messages are sent (default %default)")
-parser.add_option("-t", "--timeout", type="float", default=5,
- help="Give up after this time out (default %default)")
-opts, args = parser.parse_args()
-
-url = Url(opts.address)
-client = SyncRequestResponse(BlockingConnection(url, timeout=opts.timeout), url.path)
-
-try:
- REQUESTS= ["Twas brillig, and the slithy toves",
- "Did gire and gymble in the wabe.",
- "All mimsy were the borogroves,",
- "And the mome raths outgrabe."]
- for request in REQUESTS:
- response = client.call(Message(body=request))
- print "%s => %s" % (request, response.body)
-finally:
- client.connection.close()
+class SyncRequestClient(IncomingMessageHandler):
+ """
+ Implementation of the synchronous request-responce (aka RPC) pattern.
+ Create an instance and call invoke() to send a request and wait for a response.
+ """
+ def __init__(self, url, timeout=None):
+ """
+ @param url: a proton.Url or a URL string of the form 'host:port/path'
+ host:port is used to connect, path is used to identify the remote messaging endpoint.
+ """
+ super(SyncRequestClient, self).__init__()
+ self.connection = BlockingConnection(Url(url).defaults(), timeout=timeout)
+ self.sender = self.connection.create_sender(url.path)
+ # dynamic=true generates a unique address dynamically for this receiver.
+ # credit=1 because we want to receive 1 response message initially.
+ self.receiver = self.connection.create_receiver(None, dynamic=True, credit=1, handler=self)
+ self.response = None
+
+ def invoke(self, request):
+ """Send a request, wait for and return the response"""
+ request.reply_to = self.reply_to
+ self.sender.send(request)
+ self.connection.wait(lambda: self.response, msg="Waiting for response")
+ response = self.response
+ self.response = None # Ready for next response.
+ self.receiver.flow(1) # Set up credit for the next response.
+ return response
+
+ @property
+ def reply_to(self):
+ """Return the dynamic address of our receiver."""
+ return self.receiver.remote_source.address
+
+ def on_message(self, event):
+ """Called when we receive a message for our receiver."""
+ self.response = event.message # Store the response
+
+ def close(self):
+ self.connection.close()
+
+
+if __name__ == '__main__':
+ url = Url("0.0.0.0/examples")
+ if len(sys.argv) > 1: url = Url(sys.argv[1])
+
+ invoker = SyncRequestClient(url, timeout=2)
+ try:
+ REQUESTS= ["Twas brillig, and the slithy toves",
+ "Did gire and gymble in the wabe.",
+ "All mimsy were the borogroves,",
+ "And the mome raths outgrabe."]
+ for request in REQUESTS:
+ response = invoker.invoke(Message(body=request))
+ print "%s => %s" % (request, response.body)
+ finally:
+ invoker.close()
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/14bdfbe0/proton-c/bindings/python/proton/utils.py
----------------------------------------------------------------------
diff --git a/proton-c/bindings/python/proton/utils.py b/proton-c/bindings/python/proton/utils.py
index c15a77c..25f6ff8 100644
--- a/proton-c/bindings/python/proton/utils.py
+++ b/proton-c/bindings/python/proton/utils.py
@@ -221,7 +221,7 @@ class BlockingConnection(Handler):
while self.container.process(): pass
def wait(self, condition, timeout=False, msg=None):
- """Call process until condition() is true"""
+ """Call do_work until condition() is true"""
if timeout is False:
timeout = self.timeout
if timeout is None:
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/14bdfbe0/proton-c/include/proton/reactor.h
----------------------------------------------------------------------
diff --git a/proton-c/include/proton/reactor.h b/proton-c/include/proton/reactor.h
index 27e8fe4..36ee336 100644
--- a/proton-c/include/proton/reactor.h
+++ b/proton-c/include/proton/reactor.h
@@ -79,15 +79,6 @@ PN_EXTERN pn_connection_t *pn_reactor_connection(pn_reactor_t *reactor, pn_handl
PN_EXTERN int pn_reactor_wakeup(pn_reactor_t *reactor);
PN_EXTERN void pn_reactor_start(pn_reactor_t *reactor);
PN_EXTERN bool pn_reactor_quiesced(pn_reactor_t *reactor);
-
-/**
- * Process any available events and return. If there are no events available,
- * wait up to the timeout for more to arrive.
- *
- *@return True if there may be more events to process in future (there are still open
- * connection, timers, etc.) False if there is nothing registered with the reactor
- * that can produce further events.
- */
PN_EXTERN bool pn_reactor_process(pn_reactor_t *reactor);
PN_EXTERN void pn_reactor_stop(pn_reactor_t *reactor);
PN_EXTERN void pn_reactor_run(pn_reactor_t *reactor);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/14bdfbe0/proton-c/src/reactor/reactor.c
----------------------------------------------------------------------
diff --git a/proton-c/src/reactor/reactor.c b/proton-c/src/reactor/reactor.c
index dde3766..cd709b1 100644
--- a/proton-c/src/reactor/reactor.c
+++ b/proton-c/src/reactor/reactor.c
@@ -421,11 +421,6 @@ bool pn_reactor_process(pn_reactor_t *reactor) {
}
}
}
- // If the next event will quiesce the reactor, then give the thread back
- // before we process it so the application's processing thread can deal with
- // the effects of events processed so far.
- if (pn_reactor_quiesced(reactor))
- return true;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/3] qpid-proton git commit: PROTON-825: SyncRequestResponse should
yield in on_message.
Posted by ac...@apache.org.
PROTON-825: SyncRequestResponse should yield in on_message.
This is the replacement fix for the reverted commit:
1455f32 PROTON-825: Reactor should yield before quiescing.
Instead of changing the reactor process() logic, we call pn_reactor_yield
from on_message in the SyncRequestResponse handler.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/2f854430
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/2f854430
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/2f854430
Branch: refs/heads/master
Commit: 2f854430c1490aaaf7e5a9545921a42b8454cf68
Parents: 14bdfbe
Author: Alan Conway <ac...@redhat.com>
Authored: Tue Feb 24 09:17:56 2015 -0500
Committer: Alan Conway <ac...@redhat.com>
Committed: Tue Feb 24 09:21:39 2015 -0500
----------------------------------------------------------------------
proton-c/bindings/python/proton/utils.py | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/2f854430/proton-c/bindings/python/proton/utils.py
----------------------------------------------------------------------
diff --git a/proton-c/bindings/python/proton/utils.py b/proton-c/bindings/python/proton/utils.py
index 25f6ff8..d5e2e0a 100644
--- a/proton-c/bindings/python/proton/utils.py
+++ b/proton-c/bindings/python/proton/utils.py
@@ -221,7 +221,7 @@ class BlockingConnection(Handler):
while self.container.process(): pass
def wait(self, condition, timeout=False, msg=None):
- """Call do_work until condition() is true"""
+ """Call process until condition() is true"""
if timeout is False:
timeout = self.timeout
if timeout is None:
@@ -329,3 +329,4 @@ class SyncRequestResponse(IncomingMessageHandler):
def on_message(self, event):
"""Called when we receive a message for our receiver."""
self.response = event.message
+ self.connection.container.yield_() # Wake up the wait() loop to handle the message.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[3/3] qpid-proton git commit: NO-JIRA: Use SyncRequestResponse in
sync_client.py example
Posted by ac...@apache.org.
NO-JIRA: Use SyncRequestResponse in sync_client.py example
Also minor improvements to server.py and server_direct.py examples.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/dc27a7b1
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/dc27a7b1
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/dc27a7b1
Branch: refs/heads/master
Commit: dc27a7b1cc73031fdd256e63f20ada80aaa60073
Parents: 2f85443
Author: Alan Conway <ac...@redhat.com>
Authored: Tue Feb 24 08:49:16 2015 -0500
Committer: Alan Conway <ac...@redhat.com>
Committed: Tue Feb 24 09:21:39 2015 -0500
----------------------------------------------------------------------
examples/python/server.py | 13 +++---
examples/python/server_direct.py | 7 ++-
examples/python/sync_client.py | 80 ++++++++++-------------------------
3 files changed, 36 insertions(+), 64 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/dc27a7b1/examples/python/server.py
----------------------------------------------------------------------
diff --git a/examples/python/server.py b/examples/python/server.py
index fc9ac8d..63e5bbc 100755
--- a/examples/python/server.py
+++ b/examples/python/server.py
@@ -23,14 +23,15 @@ from proton.handlers import MessagingHandler
from proton.reactor import Container
class Server(MessagingHandler):
- def __init__(self, host, address):
+ def __init__(self, url, address):
super(Server, self).__init__()
- self.host = host
+ self.url = url
self.address = address
def on_start(self, event):
+ print "Listening on", self.url
self.container = event.container
- self.conn = event.container.connect(self.host)
+ self.conn = event.container.connect(self.url)
self.receiver = event.container.create_receiver(self.conn, self.address)
self.senders = {}
self.relay = None
@@ -40,16 +41,18 @@ class Server(MessagingHandler):
self.relay = self.container.create_sender(self.conn, None)
def on_message(self, event):
+ print "Received", event.message
sender = self.relay
if not sender:
sender = self.senders.get(event.message.reply_to)
if not sender:
sender = self.container.create_sender(self.conn, event.message.reply_to)
self.senders[event.message.reply_to] = sender
- sender.send(Message(address=event.message.reply_to, body=event.message.body.upper()))
+ sender.send(Message(address=event.message.reply_to, body=event.message.body.upper(),
+ correlation_id=event.message.correlation_id))
try:
- Container(Server("localhost:5672", "examples")).run()
+ Container(Server("0.0.0.0:5672", "examples")).run()
except KeyboardInterrupt: pass
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/dc27a7b1/examples/python/server_direct.py
----------------------------------------------------------------------
diff --git a/examples/python/server_direct.py b/examples/python/server_direct.py
index 605c755..25af702 100755
--- a/examples/python/server_direct.py
+++ b/examples/python/server_direct.py
@@ -29,6 +29,7 @@ class Server(MessagingHandler):
self.senders = {}
def on_start(self, event):
+ print "Listening on", self.url
self.container = event.container
self.acceptor = event.container.listen(self.url)
@@ -46,12 +47,14 @@ class Server(MessagingHandler):
event.link.target.address = event.link.remote_target.address
def on_message(self, event):
+ print "Received", event.message
sender = self.senders.get(event.message.reply_to)
if sender:
- sender.send(Message(address=event.message.reply_to, body=event.message.body.upper()))
+ sender.send(Message(address=event.message.reply_to, body=event.message.body.upper(),
+ correlation_id=event.message.correlation_id))
try:
- Container(Server("localhost:8888")).run()
+ Container(Server("0.0.0.0:8888")).run()
except KeyboardInterrupt: pass
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/dc27a7b1/examples/python/sync_client.py
----------------------------------------------------------------------
diff --git a/examples/python/sync_client.py b/examples/python/sync_client.py
index 86cd7c2..82cd85f 100755
--- a/examples/python/sync_client.py
+++ b/examples/python/sync_client.py
@@ -24,65 +24,31 @@ Demonstrates the client side of the synchronous request-response pattern
"""
+import optparse
from proton import Message, Url, ConnectionException, Timeout
-from proton.utils import BlockingConnection
+from proton.utils import SyncRequestResponse, BlockingConnection
from proton.handlers import IncomingMessageHandler
import sys
-class SyncRequestClient(IncomingMessageHandler):
- """
- Implementation of the synchronous request-responce (aka RPC) pattern.
- Create an instance and call invoke() to send a request and wait for a response.
- """
+parser = optparse.OptionParser(usage="usage: %prog [options]",
+ description="Send requests to the supplied address and print responses.")
+parser.add_option("-a", "--address", default="localhost:5672/examples",
+ help="address to which messages are sent (default %default)")
+parser.add_option("-t", "--timeout", type="float", default=5,
+ help="Give up after this time out (default %default)")
+opts, args = parser.parse_args()
+
+url = Url(opts.address)
+client = SyncRequestResponse(BlockingConnection(url, timeout=opts.timeout), url.path)
+
+try:
+ REQUESTS= ["Twas brillig, and the slithy toves",
+ "Did gire and gymble in the wabe.",
+ "All mimsy were the borogroves,",
+ "And the mome raths outgrabe."]
+ for request in REQUESTS:
+ response = client.call(Message(body=request))
+ print "%s => %s" % (request, response.body)
+finally:
+ client.connection.close()
- def __init__(self, url, timeout=None):
- """
- @param url: a proton.Url or a URL string of the form 'host:port/path'
- host:port is used to connect, path is used to identify the remote messaging endpoint.
- """
- super(SyncRequestClient, self).__init__()
- self.connection = BlockingConnection(Url(url).defaults(), timeout=timeout)
- self.sender = self.connection.create_sender(url.path)
- # dynamic=true generates a unique address dynamically for this receiver.
- # credit=1 because we want to receive 1 response message initially.
- self.receiver = self.connection.create_receiver(None, dynamic=True, credit=1, handler=self)
- self.response = None
-
- def invoke(self, request):
- """Send a request, wait for and return the response"""
- request.reply_to = self.reply_to
- self.sender.send(request)
- self.connection.wait(lambda: self.response, msg="Waiting for response")
- response = self.response
- self.response = None # Ready for next response.
- self.receiver.flow(1) # Set up credit for the next response.
- return response
-
- @property
- def reply_to(self):
- """Return the dynamic address of our receiver."""
- return self.receiver.remote_source.address
-
- def on_message(self, event):
- """Called when we receive a message for our receiver."""
- self.response = event.message # Store the response
-
- def close(self):
- self.connection.close()
-
-
-if __name__ == '__main__':
- url = Url("0.0.0.0/examples")
- if len(sys.argv) > 1: url = Url(sys.argv[1])
-
- invoker = SyncRequestClient(url, timeout=2)
- try:
- REQUESTS= ["Twas brillig, and the slithy toves",
- "Did gire and gymble in the wabe.",
- "All mimsy were the borogroves,",
- "And the mome raths outgrabe."]
- for request in REQUESTS:
- response = invoker.invoke(Message(body=request))
- print "%s => %s" % (request, response.body)
- finally:
- invoker.close()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org