Handle messages received on the socket. Some messages received may be asynchronous 'call' or 'queue' requests, and some may be responses for other threads. 'call' requests are passed to self.localcall() with the expectation of immediate execution, during which time
(self, myseq, wait)
| 395 | return message |
| 396 | |
| 397 | def pollresponse(self, myseq, wait): |
| 398 | """Handle messages received on the socket. |
| 399 | |
| 400 | Some messages received may be asynchronous 'call' or 'queue' requests, |
| 401 | and some may be responses for other threads. |
| 402 | |
| 403 | 'call' requests are passed to self.localcall() with the expectation of |
| 404 | immediate execution, during which time the socket is not serviced. |
| 405 | |
| 406 | 'queue' requests are used for tasks (which may block or hang) to be |
| 407 | processed in a different thread. These requests are fed into |
| 408 | request_queue by self.localcall(). Responses to queued requests are |
| 409 | taken from response_queue and sent across the link with the associated |
| 410 | sequence numbers. Messages in the queues are (sequence_number, |
| 411 | request/response) tuples and code using this module removing messages |
| 412 | from the request_queue is responsible for returning the correct |
| 413 | sequence number in the response_queue. |
| 414 | |
| 415 | pollresponse() will loop until a response message with the myseq |
| 416 | sequence number is received, and will save other responses in |
| 417 | self.responses and notify the owning thread. |
| 418 | |
| 419 | """ |
| 420 | while True: |
| 421 | # send queued response if there is one available |
| 422 | try: |
| 423 | qmsg = response_queue.get(0) |
| 424 | except queue.Empty: |
| 425 | pass |
| 426 | else: |
| 427 | seq, response = qmsg |
| 428 | message = (seq, ('OK', response)) |
| 429 | self.putmessage(message) |
| 430 | # poll for message on link |
| 431 | try: |
| 432 | message = self.pollmessage(wait) |
| 433 | if message is None: # socket not ready |
| 434 | return None |
| 435 | except EOFError: |
| 436 | self.handle_EOF() |
| 437 | return None |
| 438 | except AttributeError: |
| 439 | return None |
| 440 | seq, resq = message |
| 441 | how = resq[0] |
| 442 | self.debug("pollresponse:%d:myseq:%s" % (seq, myseq)) |
| 443 | # process or queue a request |
| 444 | if how in ("CALL", "QUEUE"): |
| 445 | self.debug("pollresponse:%d:localcall:call:" % seq) |
| 446 | response = self.localcall(seq, resq) |
| 447 | self.debug("pollresponse:%d:localcall:response:%s" |
| 448 | % (seq, response)) |
| 449 | if how == "CALL": |
| 450 | self.putmessage((seq, response)) |
| 451 | elif how == "QUEUE": |
| 452 | # don't acknowledge the 'queue' request! |
| 453 | pass |
| 454 | continue |
no test coverage detected