Rev pulsar client code.

Updates Pulsar client through galaxyproject/pulsar@f1a9e83. The most important and substantial change here is a fix for galaxyproject/pulsar#63 provided by @dctrud.
galaxyproject · Apr 10, 2015 · aad36d0 · aad36d0
1 parent 800e83f
commit aad36d0
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 4 deletions.
diff --git a/lib/pulsar/client/amqp_exchange.py b/lib/pulsar/client/amqp_exchange.py
@@ -83,7 +83,7 @@ def consume(self, queue_name, callback, check=True, connection_kwargs={}):
                                 connection.drain_events(timeout=self.__timeout)
                             except socket.timeout:
                                 pass
-            except (IOError, socket.error), exc:
+            except (IOError, socket.error) as exc:
                 self.__handle_io_error(exc, heartbeat_thread)
             except BaseException:
                 log.exception("Problem consuming queue, consumer quitting in problematic fashion!")

diff --git a/lib/pulsar/client/client.py b/lib/pulsar/client/client.py
@@ -1,5 +1,6 @@
 import os
 from json import dumps
+from json import loads
 
 from .destination import submit_params
 from .setup_handler import build as build_setup_handler
@@ -170,10 +171,13 @@ def put_file(self, path, input_type, name=None, contents=None, action_type='tran
         input_path = path
         if contents:
             input_path = None
-        if action_type == 'transfer':
+        # action type == 'message' should either copy or transfer
+        # depending on default not just fallback to transfer.
+        if action_type in ['transfer', 'message']:
             return self._upload_file(args, contents, input_path)
         elif action_type == 'copy':
-            pulsar_path = self._raw_execute('path', args)
+            path_response = self._raw_execute('path', args)
+            pulsar_path = loads(path_response)['path']
             copy(path, pulsar_path)
             return {'path': pulsar_path}
 

diff --git a/lib/pulsar/client/staging/up.py b/lib/pulsar/client/staging/up.py
@@ -348,7 +348,7 @@ def handle_transfer(self, path, type, name=None, contents=None):
         if action.staging_needed:
             local_action = action.staging_action_local
             if local_action:
-                response = self.client.put_file(path, type, name=name, contents=contents)
+                response = self.client.put_file(path, type, name=name, contents=contents, action_type=action.action_type)
 
                 def get_path():
                     return response['path']