Fix a few bugs to get the basic example working

[chadrik]

Hi @jonnor, it's me again.
The repeat.py example now properly repeats data without raising an exception, but it immediately shuts down after forwarding a single message. I'm going to look into that next unless you know what's going on. It prints this:

Publising message: {u'foo': u'1111'}, out, [{'queue': 'repeat.OUT', 'type': 'any', 'id': 'out'}]
AMQP channel closed; close-info: {'class_id': 60, 'method_id': 80, 'reply_code': 406, 'reply_text': 'PRECONDITION_FAILED - unknown delivery tag 1'}
AMQP broker connection closed; close-info: {'class_id': 0, 'method_id': 0, 'reply_code': 0, 'reply_text': ''}
Shutdown

[jonnor]

Hi @chadrik! What was the exception you got before? The tests passed before as well so I find it a bit strange that change was neccesary (unless Haigha changed API?)...

Also, the AMQP error P RECONDITION_FAILED - unknown delivery tag 1 suggests either ACK/NACK-ing a message twice, or ACK/NACKing a message that was not sent to us.

[jonnor]

Just re-ran the test of version without this fix on Travis, also passed.

[chadrik]

I haven't had time to look through the tests, but I suspect that the problem lies there. I see that the tests are running repeat.py, but if Repeat.process() gets called an error will be raised when it calls self.ack(). The problem I fixed is pretty fundamental: AttributeError: 'Channel' object has no attribute 'ack'. Here's the full traceback:

Traceback (most recent call last):
  File "/Library/Python/2.7/site-packages/gevent/greenlet.py", line 327, in run
    result = self._run(*self.args, **self.kwargs)
  File "/Users/chad/dev/noflo-dev/msgflo-python/msgflo.py", line 122, in _message_pump_greenthread
    self._conn.read_frames()
  File "/Library/Python/2.7/site-packages/haigha/connection.py", line 445, in read_frames
    self._transport.process_channels(p_channels)
  File "/Library/Python/2.7/site-packages/haigha/transports/transport.py", line 40, in process_channels
    channel.process_frames()
  File "/Library/Python/2.7/site-packages/haigha/channel.py", line 234, in process_frames
    self.dispatch(frame)
  File "/Library/Python/2.7/site-packages/haigha/channel.py", line 212, in dispatch
    klass.dispatch(method_frame)
  File "/Library/Python/2.7/site-packages/haigha/classes/protocol_class.py", line 80, in dispatch
    callback(method_frame)
  File "/Library/Python/2.7/site-packages/haigha/classes/basic_class.py", line 242, in _recv_deliver
    func(msg)
  File "/Users/chad/dev/noflo-dev/msgflo-python/msgflo.py", line 67, in handleInput
    part.process(port, msg)
  File "./examples/repeat.py", line 19, in process
    self.ack(msg)
  File "/Users/chad/dev/noflo-dev/msgflo-python/msgflo.py", line 48, in ack
    self._runtime._channel.ack(msg.delivery_tag)
AttributeError: 'Channel' object has no attribute 'ack'
<Greenlet at 0x1010e4d70: <bound method GeventEngine._message_pump_greenthread of <msgflo.GeventEngine object at 0x1010d5dd0>>> failed with AttributeError

The error I got after fixing that was AttributeError: 'Message' object has no attribute 'delivery_tag' on the same lines. I thought it must be a version issue as well, but I'm using the same versions that are in the requirements.pip. So if the versions are the same and the tests are running repeat.py, I don't see any way that the test is working. One thing I noticed, however, is that the return status of the python process after the exception is 0, which could be tricking the tests into thinking they succeeded. I couldn't even drop into pdb properly after the exception. It must be something about the threaded gevent loop. I'm going to do a little research to see how exceptions should be propagated up to the main thread.

The way I'm testing is this:

start the graph

echo "repeat(Repeat) OUT -> IN out(Output)" > myservice.fbp
msgflo-setup ./myservice.fbp --broker amqp://localhost

setup the output node

noflo-runtime-msgflo --name out --graph core/Output --broker amqp://localhost

setup the repeat node

python ./examples/repeat.py

send a message

msgflo-send-message --queue repeat.IN --json '{ "foo": "1111" }'

[chadrik]

hey @jonnor, I had a co-worker (@Flushot) suss out the double-ack issue. I think everything should be working properly now.

[jonnor]

Merged. Thanks @chadrik and @Flushot !