pika/pika — Issue #122: Pika cannot handle tcpkill command. 'NoneType' is not catched in Pika library - particularly in File "/usr/local/lib/python2.7/dist-packages/pika-0.9.5-py2.7.egg/pika/adapters/base_connection.py", in function "def _handle_error(self, error):"
Author: lockonoko
Created: 2012-02-10 19:46:13+00:00
State: open

I wrote the python program to send and receive messages to/from RabbitMq Server as followed:
In main function:
if __name__ == '__main__':
def post_heartbeat():
heartbeat(d2l_agent)
Timer(300.0, post_heartbeat).start()	# Send a 'heartbeat' to the queue every 30 minutes

post_heartbeat()

fetch_job(d2l_agent)
exit(0)

***********************************************************************************************************************************
heartbeat(_) is a global function to send messages to amq server. It is defined in d2lAgentPublish.py as followed:

from pika.adapters import SelectConnection
from pika.reconnection_strategies import SimpleReconnectionStrategy
import pika
import sys
import traceback
import time
'''
Continuation passing style setup for RabbitMQ initialization
'''
# Create a global channel variable to hold our channel object in
channel = agent = None

# Step #2
def on_connected(connection):
"""Called when we are fully connected to RabbitMQ"""
# Open a channel
connection.channel(on_channel_open)

# Step #3
def on_channel_open(new_channel):
"""Called when our channel has opened"""
global channel
channel = new_channel
channel.basic_qos(prefetch_count = 1)
channel.queue_declare(queue=agent.mq_queue, durable=True, exclusive=False, auto_delete=False, callback=on_queue_declared)

# Step #4
def on_queue_declared(frame):
"""Called when RabbitMQ has told us our Queue has been declared, frame is the response from RabbitMQ"""
channel.basic_consume(on_message_receipt, queue=agent.mq_queue)

# Step #5

def on_message_receipt(channel, method, header, job_uuid):
try:
agent.log.info( '(): queue: ' + agent.mq_queue + ', received body: ' + str(job_uuid) )

if (str(job_uuid).startswith('heartbeat')):
agent.log.info( "(): ACKing: " + str(job_uuid) )
channel.basic_ack(delivery_tag = method.delivery_tag)
time.sleep(15)
return	# Ignore heartbeats

agent.log.info( "on_message_receipt(): received job: " + job_uuid )

if agent.action == 'provision':
if agent.d2l_provision( job_uuid):
agent.log.info( "on_message_receipt(): acked provision job: " + job_uuid )
channel.basic_ack(delivery_tag = method.delivery_tag)
else:
agent.log.info( "on_message_receipt(): nacked provision job: " + job_uuid )
channel.basic_reject(delivery_tag = method.delivery_tag, requeue = True)	# Return job to the queue
else:
if agent.d2l_chpasswd(job_uuid):
channel.basic_ack(delivery_tag = method.delivery_tag)
agent.log.info( "on_message_receipt(): acked password job: " + job_uuid )
else:
channel.basic_reject(delivery_tag = method.delivery_tag, requeue = True)	# Return job to the queue
agent.log.info( "on_message_receipt(): nacked password job: " + job_uuid )

agent.log.info( "on_message_receipt(): finished processing job: " + job_uuid )
except Exception as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
agent.log.error( "on_message_receipt(): ERROR: Received exception: " + str(e)
+ repr(traceback.format_exception(exc_type, exc_value, exc_traceback)) )

def rough_connect(parameters,on_connected,srs):
connection = SelectConnection(parameters, on_open_callback=on_connected, reconnection_strategy=srs)

try:
# Loop so we can communicate with RabbitMQ
connection.ioloop.start()
except KeyboardInterrupt:
# Gracefully close the connection
connection.close()
# Loop until we're fully closed, will stop on its own
connection.ioloop.start()
#	except Exception as inst:
#	x = inst
#	print "Encounter exception " + str(x) + " create another connection"
#	rough_connect(parameters,on_connected,srs)

def fetch_job(d2l_agent):
global agent
agent = d2l_agent
agent.log.info('post_job(): MQ Parameter setup')
credentials = pika.PlainCredentials( agent.mq_username, agent.mq_password)

parameters = pika.ConnectionParameters(
host = agent.mq_hostname,
credentials = credentials,
virtual_host = agent.mq_vhost)
#heartbeat = True

agent.log.info('post_job(): Setting-up connection, host: ' + agent.mq_hostname )

# Step #1: Connect to RabbitMQ
srs = SimpleReconnectionStrategy()
#rough_connect(parameters,on_connected,srs)
connection = SelectConnection(parameters, on_open_callback=on_connected, reconnection_strategy=srs)
try:
# Loop so we can communicate with RabbitMQ
connection.ioloop.start()
except KeyboardInterrupt:
# Gracefully close the connection
connection.close()
# Loop until we're fully closed, will stop on its own
connection.ioloop.start()