Handling Kafka and NATS message queue trigger

@param {MQTypes} mqType
@param {string} name
@param {string} namespace
@param {string} topic
@param {object} options
764499
function deployMQTrigger(mqType, name, namespace, topic, options) {
  const opts = _.defaults({}, options, {
    log: console.log,
  });
  const trigger = {
    apiVersion: 'kubeless.io/v1beta1',
    kind: `${mqType}Trigger`,
    metadata: {
      name: _.kebabCase(`${name}-${topic}`),
      namespace,
      labels: {
        'created-by': 'kubeless',
      },
    },
    spec: {
      functionSelector: {
        matchLabels: {
          'created-by': 'kubeless',
          function: name,
        },
      },
      topic,
    },
  };
  const triggerApi = new CRD(
    'apis/kubeless.io',
    'v1beta1',
    namespace,
    `${mqType.toLowerCase()}triggers`
  );
  opts.log(`Deploying ${mqType} trigger: ${trigger.metadata.name}`);
  return triggerApi.getItem(trigger.metadata.name)
    .then((res) => {
      if (res.code === 404) {
        return triggerApi.post({ body: trigger });
      }
      opts.log(`Updating existing ${mqType} trigger`);
      return triggerApi.put(trigger.metadata.name, { body: trigger });
    });
}
212348
public void setTopicSpaceName(String name)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setTopicSpaceName", name);

        topicSpaceName = name;

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "setTopicSpaceName");
    }
921296
public <T> QueuePublisher<T> createQueuePublisher(final String name, final PublisherCallback<T> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new QueuePublisher<T>(connectionFactory, amqpConfig, name, messageCallback);
    }
456966
@RobotKeywordOverload
    @ArgumentNames({ "topic", "message" })
    public void publishToMQTTSynchronously(String topic, Object message)
            throws MqttException {
        publishToMQTTSynchronously(topic, message, 0, false);
    }
1039295
function Subscription(name, topic) {
            this._pattern = "%s://%s.mns.%s.aliyuncs.com/topics/%s/subscriptions/%s";
            this._name = name;
            this._topic = topic;
            // make url
            this._urlAttr = this.makeAttrURL();
            // create the OpenStack object
            this._openStack = new AliMNS.OpenStack(topic.getAccount());
        }
656981
def publish!(topic, subtopic, schema, partition_key, payload)
      event = Event.new(topic, subtopic, origin, schema, partition_key, payload)
      raise MegaphoneInvalidEventError.new(event.errors.join(', ')) unless event.valid?
      unless logger.post(topic, event.to_hash)
        if transient_error?(logger.last_error)
          raise MegaphoneMessageDelayWarning.new(logger.last_error.message, event.stream_id)
        else
          raise MegaphoneUnavailableError.new(logger.last_error.message, event.stream_id)
        end
      end
    end
139814
@Override
    public MessageConsumer createDurableConsumer(Topic topic,
                                                 String name)
                    throws IllegalStateException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createDurableConsumer", new Object[] { topic, name });
        throw (javax.jms.IllegalStateException) JmsErrorUtils.newThrowable(javax.jms.IllegalStateException.class,
                                                                           "INVALID_OP_FOR_CLASS_CWSIA0481",
                                                                           new Object[] { "createDurableConsumer", "QueueSession" },
                                                                           tc);

    }
1253657
public function subscribe($name, $id = null, $callback = null, $timeout = null, $failswitch = false) {
        $self = $this;
        if (!is_string($name)) {
            throw new TypeError('topic name must be a string');
        }
        if (is_callable($id) && !is_callable($callback)) {
            $timeout = $callback;
            $callback = $id;
            $id = null;
        }
        if (!is_callable($callback)) {
            throw new TypeError('callback must be a function.');
        }
        if (!isset($this->topics[$name])) {
            $this->topics[$name] = array();
        }
        if ($id === null) {
            if ($this->id == null) {
                $this->id = $this->autoId();
            }
            $this->id->then(function($id) use ($self, $name, $callback, $timeout, $failswitch) {
                $self->;
            });
            return;
        }
        // Default subscribe timeout is 5 minutes.
        if (!is_int($timeout)) $timeout = 300000;
        $topic = $this->getTopic($name, $id);
        if ($topic === null) {
            $topic = new stdClass();
            $settings = new InvokeSettings(array(
                'idempotent' => true,
                'failswitch' => $failswitch,
                'timeout' => $timeout
            ));
            $cb = function() use ($self, &$cb, $topic, $name, $id, $settings) {
                $args = array($id);
                $self->invoke($name, $args, $settings)
                     ->then($topic->handler, $cb);
            };
            $topic->handler = function($result) use ($self, $name, $id, $cb) {
                $topic = $self->getTopic($name, $id);
                if ($topic !== null) {
                    if ($result !== null) {
                        $callbacks = $topic->callbacks;
                        foreach ($callbacks as $callback) {
                            try {
                                call_user_func($callback, $result);
                            }
                            catch (Exception $ex) {}
                            catch (Throwable $ex) {}
                        }
                    }
                    if ($self->getTopic($name, $id) !== null) $cb();
                }
            };
            $topic->callbacks = array($callback);
            $this->topics[$name][$id] = $topic;
            $cb();
        }
        elseif (array_search($callback, $topic->callbacks, true) === false) {
            $topic->callbacks[] = $callback;
        }
    }
1193686
function Question(name, message, options) {
  if (utils.isObject(name)) {
    options = utils.merge({}, message, name);
    message = options.message;
    name = options.name;
  }

  if (utils.isObject(message)) {
    options = utils.merge({}, options, message);
    message = options.message;
  }

  utils.define(this, 'isQuestion', true);
  utils.define(this, 'cache', {});

  this.options = options || {};
  this.type = this.options.type || 'input';
  this.message = message || this.options.message;
  this.name = name || this.options.name;

  if (!this.message) {
    this.message = this.name;
  }

  utils.merge(this, this.options);
  createNext(this);
}
573316
public boolean removeMessageListener(String topic, IKafkaMessageListener msgListener) {
        synchronized (topicMsgListeners) {
            if (topicMsgListeners.remove(topic, msgListener)) {
                if (topicMsgListeners.get(topic).isEmpty()) {
                    // no more listener, stop worker
                    KafkaMsgConsumerWorker worker = topicWorkers.remove(topic);
                    if (worker != null) {
                        worker.stopWorker();
                    }
                }
                return true;
            }
        }
        return false;
    }
638067
private Controllable createTopicMBean(Controllable c) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createPublicationPointMBean", new Object[] { c });
        TopicImpl pp = new TopicImpl(_me, c);
        controllableMap.put(c, pp);
        try {
            Dictionary<String, Object> properties = new Hashtable<String, Object>();
            properties.put("service.vendor", "IBM");
            String cName = c.getName();
            properties.put("jmx.objectname", "WebSphere:feature=wasJmsServer,type=Topic,name=" + cName);
            ServiceRegistration<TopicMBean> topicMBean = (ServiceRegistration<TopicMBean>) bcontext.registerService(TopicMBean.class.getName(), pp, properties);
            serviceObjects.put(cName, topicMBean);
        } catch (Exception e) {
            SibTr.exception(tc, e);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createPublicationPointMBean", new Object[] { pp });
        return pp;

    }
547279
function NodeMqttTransport(options) {
  Transport.call(this, options);

  this._options = options;
  this._client = null;
  this._sendTimer = null;
  this._buf = [];

  this._status = '';

  this._connHandler = onConnect.bind(this);
  this._messageHandler = onMessage.bind(this);
  this._sendOutHandler = sendOut.bind(this);
  this._disconnHandler = onDisconnect.bind(this);
  this._errorHandler = onError.bind(this);

  init(this);
}
129102
def create_async_sns_topic(self, lambda_name, lambda_arn):
        
        topic_name = get_topic_name(lambda_name)
        # Create SNS topic
        topic_arn = self.sns_client.create_topic(
            Name=topic_name)['TopicArn']
        # Create subscription
        self.sns_client.subscribe(
            TopicArn=topic_arn,
            Protocol='lambda',
            Endpoint=lambda_arn
        )
        # Add Lambda permission for SNS to invoke function
        self.create_event_permission(
            lambda_name=lambda_name,
            principal='sns.amazonaws.com',
            source_arn=topic_arn
        )
        # Add rule for SNS topic as a event source
        add_event_source(
            event_source={
                "arn": topic_arn,
                "events": ["sns:Publish"]
            },
            lambda_arn=lambda_arn,
            target_function="zappa.asynchronous.route_task",
            boto_session=self.boto_session
        )
        return topic_arn
1371020
@When("^I send a message '(.+?)' to the kafka topic named '(.+?)'")
    public void sendAMessage(String message, String topic_name) throws Exception {
        commonspec.getKafkaUtils().sendMessage(message, topic_name);
    }
336708
function publish(topic, msg) {
  if (!topic || !msg || typeof topic !== 'string' || typeof msg !== 'object') {
    throw new Error('A `topic` (string) and `msg` (object) property must be defined');
  }

  if (module.exports.client && BUS_MODULE) {
    module.exports.client.publish(`${NAMESPACE}:${topic}`, msg);
  } else if (module.exports.client) {
    module.exports.client.publish(`${NAMESPACE}:${topic}`, JSON.stringify(msg));
  }
}
1095356
public function publishMeta($topicName, $arguments, $argumentsKw = null, $options = null)
    {
        if ($this->metaSession === null) {
            // setup a new metaSession
            $s                 = new Session(new DummyTransport());
            $this->metaSession = $s;
        }

        $messageEvent = new MessageEvent($this->metaSession,
            new PublishMessage(
                Utils::getUniqueId(),
                $options,
                $topicName,
                $arguments,
                $argumentsKw
            ));
        $this->getBroker()->handlePublishMessage($messageEvent);
    }
1125184
def publish(self, topic, payload, qos, retain):
        """"""
        self._mqttc.publish(topic, payload, qos, retain)