<commit_before>module Mcoflow
  module Connectors
    class MCollective < Dynflow::MicroActor

      include ::MCollective::RPC

      REPLY_QUEUE = '/queue/mcollective.mcoflow'

      Task = Algebrick.type do
        fields(suspended_action: Dynflow::Action::Suspended,
               mco_filter: Hash,
               mco_agent: Object,
               mco_action: Object,
               mco_args: Hash)
      end

      Response = Algebrick.type do
        fields(request_id:   String,
               payload:      Hash)
      end

      Event = Algebrick.type do
        fields(done: Object,
               payload: Object)
      end

      def initialize(logger)
        super(logger)
        @actions           =  {}

        stomp_connection.subscribe(REPLY_QUEUE, { :id => 1 })

        @listener = Thread.new do
          loop do
            begin
              raw_msg = stomp_connection.receive
              message  = ::MCollective::Message.new(raw_msg.body, raw_msg)
              message.type = :reply
              message.decode!
              self << Response[message.requestid, message.payload]
            rescue => e
              puts "Error #{e.message}"
              puts e.backtrace.join("\n")
            end
          end
        end
      end

      def mco_run(suspended_action, mco_filter, mco_agent, mco_action, mco_args)
        task = Task[suspended_action, mco_filter, mco_agent, mco_action, mco_args]
        self.ask(task).value!
      end

      private

      def load_stomp_config
        ::MCollective::Config.instance.pluginconf.reduce({}) do |h, (k, v)|
          (key = k[/^activemq.*\.(\w+)$/,1]) ? h.update(key => v) : h
        end
      end

      def stomp_connection
        return @connection if @connection
        stomp_config = load_stomp_config
        config = {:hosts => [{ :login => stomp_config["user"],
                               :passcode => stomp_config["password"],
                               :host => stomp_config["host"],
                               :port => stomp_config["port"] }]}
        @connection = Stomp::Connection.open(config)
      end

      def on_message(message)
        case message
        when Task
          trigger_task(message)
        when Response
          resume_action(message)
        end
      end

      def trigger_task(task)
        client = rpcclient(task[:mco_agent].to_s)
        task[:mco_filter].each do |key, value|
          client.send(key, *Array(value))
        end
        client.reply_to = REPLY_QUEUE
        request_id = client.send(task[:mco_action], task[:mco_args])<commit_msg>Fix the args handling<commit_after>module Mcoflow
  module Connectors
    class MCollective < Dynflow::MicroActor

      include ::MCollective::RPC

      REPLY_QUEUE = '/queue/mcollective.mcoflow'

      Task = Algebrick.type do
        fields(suspended_action: Dynflow::Action::Suspended,
               mco_filter: Hash,
               mco_agent: Object,
               mco_action: Object,
               mco_args: Hash)
      end

      Response = Algebrick.type do
        fields(request_id:   String,
               payload:      Hash)
      end

      Event = Algebrick.type do
        fields(done: Object,
               payload: Object)
      end

      def initialize(logger)
        super(logger)
        @actions           =  {}

        stomp_connection.subscribe(REPLY_QUEUE, { :id => 1 })

        @listener = Thread.new do
          loop do
            begin
              raw_msg = stomp_connection.receive
              message  = ::MCollective::Message.new(raw_msg.body, raw_msg)
              message.type = :reply
              message.decode!
              self << Response[message.requestid, message.payload]
            rescue => e
              puts "Error #{e.message}"
              puts e.backtrace.join("\n")
            end
          end
        end
      end

      def mco_run(suspended_action, mco_filter, mco_agent, mco_action, mco_args)
        task = Task[suspended_action, mco_filter, mco_agent, mco_action, mco_args]
        self.ask(task).value!
      end

      private

      def load_stomp_config
        ::MCollective::Config.instance.pluginconf.reduce({}) do |h, (k, v)|
          (key = k[/^activemq.*\.(\w+)$/,1]) ? h.update(key => v) : h
        end
      end

      def stomp_connection
        return @connection if @connection
        stomp_config = load_stomp_config
        config = {:hosts => [{ :login => stomp_config["user"],
                               :passcode => <PASSWORD>["password"],
                               :host => stomp_config["host"],
                               :port => stomp_config["port"] }]}
        @connection = Stomp::Connection.open(config)
      end

      def on_message(message)
        case message
        when Task
          trigger_task(message)
        when Response
          resume_action(message)
        end
      end

      def trigger_task(task)
        client = rpcclient(task[:mco_agent].to_s,
                           options: ::MCollective::Util.default_options)
        task[:mco_filter].each do |key, value|
          client.send(key, *Array(value))
        end
        client.reply_to = REPLY_QUEUE
        request_id = client.send(task[:mco_action], task[:mco_args])
{'fraction_non_alphanumeric': 0.09986734887246541, 'fraction_numerical': 0.0007580064430547659, 'mean_word_length': 1.7011258955987718, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5815282', 'n_tokens_mistral': 1552, 'n_tokens_neox': 1511, 'n_words': 312}