lib/actions/pub.js
/**
 * This method will be called from elastic.io platform providing following data
 *
 * @param msg incoming message object that contains ``body`` with payload
 * @param cfg configuration that is account information and configuration field values
 */
This method will be called from elastic.io platform providing following data
@param msg incoming message object that contains ``body`` with payload
@param cfg configuration that is account information and configuration field values
This method will be called from elastic . io platform providing following data @param msg incoming message object that contains ` ` body ` ` with payload @param cfg configuration that is account information and configuration field values
async function processAction(msg, cfg) {
  this.logger.info('Start publish action');
  const self = this;
  const { projectId, topicName } = cfg;
  if (!pubsubClient || !topic) {
    pubsubClient = new PubSub({
      projectId,
      credentials: {
        client_email: cfg.client_email,
        private_key: cfg.private_key,
      },
    });
    topic = pubsubClient.topic(topicName);
  }
  let bufferData;
  if ((typeof msg.body === 'object' || typeof msg.body === 'function') && (msg.body !== null)) {
    bufferData = JSON.stringify(msg.body);
    if (msg.body instanceof Buffer) {
      bufferData = Buffer.from(JSON.parse(bufferData).data);
    } else {
      bufferData = Buffer.from(bufferData);
    }
  } else {
    bufferData = Buffer.from(msg.body.toString());
  }
  try {
    const messageId = await topic.publish(bufferData);
    const body = {
      messageID: [messageId],
    };
    this.logger.info('Message sent with ID:', messageId);
    const data = messages.newMessageWithBody(body);
    await self.emit('data', data);
  } catch (err) {
    this.logger.info('Oops! Error occurred');
    await self.emit('error', err);
  }
  self.logger.info('Finished execution');
  await self.emit('end');
}
async function processAction ( msg , cfg ) { this . logger . info ( 'Start publish action' ) ; const self = this ; const { projectId , topicName } = cfg ; if ( ! pubsubClient || ! topic ) { pubsubClient = new PubSub ( { projectId , credentials : { client_email : cfg . client_email , private_key : cfg . private_key , } , } ) ; topic = pubsubClient . topic ( topicName ) ; } let bufferData ; if ( ( typeof msg . body === 'object' || typeof msg . body === 'function' ) && ( msg . body !== null ) ) { bufferData = JSON . stringify ( msg . body ) ; if ( msg . body instanceof Buffer ) { bufferData = Buffer . from ( JSON . parse ( bufferData ) . data ) ; } else { bufferData = Buffer . from ( bufferData ) ; } } else { bufferData = Buffer . from ( msg . body . toString ( ) ) ; } try { const messageId = await topic . publish ( bufferData ) ; const body = { messageID : [ messageId ] , } ; this . logger . info ( 'Message sent with ID:' , messageId ) ; const data = messages . newMessageWithBody ( body ) ; await self . emit ( 'data' , data ) ; } catch ( err ) { this . logger . info ( 'Oops! Error occurred' ) ; await self . emit ( 'error' , err ) ; } self . logger . info ( 'Finished execution' ) ; await self . emit ( 'end' ) ; }
This method will be called from elastic.io platform providing following data
 @param msg incoming message object that contains ``body`` with payload
 @param cfg configuration that is account information and configuration field values
This method will be called from elastic . io platform providing following data @param msg incoming message object that contains ` ` body ` ` with payload @param cfg configuration that is account information and configuration field values
// Lazy initialization // Google PubSub expects the payload to be of type string or an instance of Buffer, ArrayBuffer, or Array or an Array-like Object // eslint-disable-next-line no-use-before-define