Add Message Actions API support (#174)

* feat: add message actions

Add Message Actions API support which allow to: add, remove and fetch previously added actions.

feat: modify fetch messages endpoint to fetch message actions

Add new arguments to fetch messages function which allow to fetch previously added actions and
message metadata.

feat: add new event handler for message actions

Add new handler which can be used to track message actions addition / removal events.

Author: parfeon

.pubnub.yml

@@ -138,7 +138,8 @@ export default function(modules, endpoint, ...args) {
   let callInstance;
   let networkingParams = { url,
     operation: endpoint.getOperation(),
-    timeout: endpoint.getRequestTimeout(modules)
+    timeout: endpoint.getRequestTimeout(modules),
+    headers: endpoint.getRequestHeaders ? endpoint.getRequestHeaders() : {}
   };
 
   outgoingParams.uuid = config.UUID;

src/core/components/endpoint.js

@@ -3,6 +3,7 @@ import {
   MessageAnnouncement,
   StatusAnnouncement,
   SignalAnnouncement,
+  MessageActionAnnouncement,
   ObjectAnnouncement,
   CallbackStruct,
   PresenceAnnouncement,
@@ -58,6 +59,12 @@ export default class {
     });
   }
 
+  announceMessageAction(announce: MessageActionAnnouncement) {
+    this._listeners.forEach((listener) => {
+      if (listener.messageAction) listener.messageAction(announce);
+    });
+  }
+
   announceUser(announce: ObjectAnnouncement) {
     this._listeners.forEach((listener) => {
       if (listener.user) listener.user(announce);

src/core/components/listener_manager.js

@@ -6,6 +6,7 @@ import ReconnectionManager from '../components/reconnection_manager';
 import DedupingManager from '../components/deduping_manager';
 import utils from '../utils';
 import {
+  MessageActionAnnouncement,
   MessageAnnouncement,
   SignalAnnouncement,
   ObjectAnnouncement,
@@ -673,6 +674,25 @@ export default class {
         } else if (message.payload.type === 'membership') {
           this._listenerManager.announceMembership(announce);
         }
+      } else if (message.messageType === 3) {
+        // this is a message action
+        let announce: MessageActionAnnouncement = {};
+        announce.channel = channel;
+        announce.subscription = subscriptionMatch;
+        announce.timetoken = publishMetaData.publishTimetoken;
+        announce.publisher = message.issuingClientId;
+
+        announce.data = {
+          messageTimetoken: message.payload.data.messageTimetoken,
+          actionTimetoken: message.payload.data.actionTimetoken,
+          type: message.payload.data.type,
+          uuid: message.issuingClientId,
+          value: message.payload.data.value,
+        };
+
+        announce.event = message.payload.event;
+
+        this._listenerManager.announceMessageAction(announce);
       } else {
         let announce: MessageAnnouncement = {};
         announce.channel = null;

src/core/components/subscription_manager.js

@@ -13,6 +13,11 @@ export default {
   PNPublishOperation: 'PNPublishOperation',
   PNSignalOperation: 'PNSignalOperation',
 
+  // Actions API
+  PNAddMessageActionOperation: 'PNAddActionOperation',
+  PNRemoveMessageActionOperation: 'PNRemoveMessageActionOperation',
+  PNGetMessageActionsOperation: 'PNGetMessageActionsOperation',
+
   // Objects API
 
   PNCreateUserOperation: 'PNCreateUserOperation',

src/core/constants/operations.js

@@ -0,0 +1,71 @@
+/* @flow */
+
+import {
+  AddMessageActionInput,
+  AddMessageActionResponse,
+  ModulesInject,
+} from '../../flow_interfaces';
+import operationConstants from '../../constants/operations';
+
+
+export function getOperation(): string {
+  return operationConstants.PNAddMessageActionOperation;
+}
+
+export function validateParams(
+  { config }: ModulesInject,
+  incomingParams: AddMessageActionInput
+) {
+  let { action, channel, messageTimetoken } = incomingParams;
+
+  if (!messageTimetoken) return 'Missing message timetoken';
+  if (!config.subscribeKey) return 'Missing Subscribe Key';
+  if (!channel) return 'Missing message channel';
+  if (!action) return 'Missing Action';
+  if (!action.value) return 'Missing Action.value';
+  if (!action.type) return 'Missing Action.type';
+  if (action.type.length > 15) return 'Action.type value exceed maximum length of 15';
+}
+
+export function usePost() {
+  return true;
+}
+
+export function postURL(
+  { config }: ModulesInject,
+  incomingParams: AddMessageActionInput
+): string {
+  let { channel, messageTimetoken } = incomingParams;
+
+  return `/v1/message-actions/${config.subscribeKey}/channel/${channel}/message/${messageTimetoken}`;
+}
+
+export function getRequestTimeout({ config }: ModulesInject) {
+  return config.getTransactionTimeout();
+}
+
+export function getRequestHeaders(): Object {
+  return { 'Content-Type': 'application/json' };
+}
+
+export function isAuthSupported() {
+  return true;
+}
+
+export function prepareParams(): Object {
+  return {};
+}
+
+export function postPayload(
+  modules: ModulesInject,
+  incomingParams: AddMessageActionInput
+): string {
+  return incomingParams.action;
+}
+
+export function handleResponse(
+  modules: ModulesInject,
+  addMessageActionResponse: Object
+): AddMessageActionResponse {
+  return { data: addMessageActionResponse.data };
+}

src/core/endpoints/actions/add_message_action.js

@@ -0,0 +1,70 @@
+/* @flow */
+
+import {
+  GetMessageActionsInput,
+  GetMessageActionsResponse,
+  ModulesInject,
+} from '../../flow_interfaces';
+import operationConstants from '../../constants/operations';
+
+
+export function getOperation(): string {
+  return operationConstants.PNGetMessageActionsOperation;
+}
+
+export function validateParams(
+  { config }: ModulesInject,
+  incomingParams: GetMessageActionsInput
+) {
+  let { channel } = incomingParams;
+
+  if (!config.subscribeKey) return 'Missing Subscribe Key';
+  if (!channel) return 'Missing message channel';
+}
+
+
+export function getURL(
+  { config }: ModulesInject,
+  incomingParams: GetMessageActionsInput
+): string {
+  let { channel } = incomingParams;
+
+  return `/v1/message-actions/${config.subscribeKey}/channel/${channel}`;
+}
+
+export function getRequestTimeout({ config }: ModulesInject) {
+  return config.getTransactionTimeout();
+}
+
+export function isAuthSupported() {
+  return true;
+}
+
+export function prepareParams(
+  modules: ModulesInject,
+  incomingParams: GetMessageActionsInput
+): Object {
+  const { limit, start, end } = incomingParams;
+  let outgoingParams: Object = {};
+
+  if (limit) outgoingParams.limit = limit;
+  if (start) outgoingParams.start = start;
+  if (end) outgoingParams.end = end;
+
+  return outgoingParams;
+}
+
+export function handleResponse(
+  modules: ModulesInject,
+  getMessageActionsResponse: Object
+): GetMessageActionsResponse {
+  /** @type GetMessageActionsResponse */
+  let response = { data: getMessageActionsResponse.data };
+
+  if (response.data.length) {
+    response.end = response.data[response.data.length - 1].actionTimetoken;
+    response.start = response.data[0].actionTimetoken;
+  }
+
+  return response;
+}

src/core/endpoints/actions/get_message_actions.js

@@ -0,0 +1,57 @@
+/* @flow */
+
+import {
+  RemoveMessageActionInput,
+  RemoveMessageActionResponse,
+  ModulesInject,
+} from '../../flow_interfaces';
+import operationConstants from '../../constants/operations';
+
+
+export function getOperation(): string {
+  return operationConstants.PNRemoveMessageActionOperation;
+}
+
+export function validateParams(
+  { config }: ModulesInject,
+  incomingParams: RemoveMessageActionInput
+) {
+  let { channel, actionTimetoken, messageTimetoken } = incomingParams;
+
+  if (!messageTimetoken) return 'Missing message timetoken';
+  if (!actionTimetoken) return 'Missing action timetoken';
+  if (!config.subscribeKey) return 'Missing Subscribe Key';
+  if (!channel) return 'Missing message channel';
+}
+
+export function useDelete() {
+  return true;
+}
+
+export function getURL(
+  { config }: ModulesInject,
+  incomingParams: RemoveMessageActionInput
+): string {
+  let { channel, actionTimetoken, messageTimetoken } = incomingParams;
+
+  return `/v1/message-actions/${config.subscribeKey}/channel/${channel}/message/${messageTimetoken}/action/${actionTimetoken}`;
+}
+
+export function getRequestTimeout({ config }: ModulesInject) {
+  return config.getTransactionTimeout();
+}
+
+export function isAuthSupported() {
+  return true;
+}
+
+export function prepareParams(): Object {
+  return {};
+}
+
+export function handleResponse(
+  modules: ModulesInject,
+  removeMessageActionResponse: Object
+): RemoveMessageActionResponse {
+  return { data: removeMessageActionResponse.data };
+}

src/core/endpoints/actions/remove_message_action.js

@@ -29,22 +29,27 @@ export function validateParams(
   modules: ModulesInject,
   incomingParams: FetchMessagesArguments
 ) {
-  let { channels } = incomingParams;
+  let { channels, includeMessageActions = false } = incomingParams;
   let { config } = modules;
 
   if (!channels || channels.length === 0) return 'Missing channels';
   if (!config.subscribeKey) return 'Missing Subscribe Key';
+
+  if (includeMessageActions && channels.length > 1) {
+    throw new TypeError('History can return actions data for a single channel only. Either pass a single channel or disable the includeMessageActions flag.');
+  }
 }
 
 export function getURL(
   modules: ModulesInject,
   incomingParams: FetchMessagesArguments
 ): string {
-  let { channels = [] } = incomingParams;
+  let { channels = [], includeMessageActions = false } = incomingParams;
   let { config } = modules;
+  const endpoint = !includeMessageActions ? 'history' : 'history-with-actions';
 
   let stringifiedChannels = channels.length > 0 ? channels.join(',') : ',';
-  return `/v3/history/sub-key/${
+  return `/v3/${endpoint}/sub-key/${
     config.subscribeKey
   }/channel/${utils.encodeString(stringifiedChannels)}`;
 }
@@ -61,13 +66,20 @@ export function prepareParams(
   modules: ModulesInject,
   incomingParams: FetchMessagesArguments
 ): Object {
-  const { start, end, count, stringifiedTimeToken = false } = incomingParams;
+  const {
+    start,
+    end,
+    count,
+    stringifiedTimeToken = false,
+    includeMeta = false,
+  } = incomingParams;
   let outgoingParams: Object = {};
 
   if (count) outgoingParams.max = count;
   if (start) outgoingParams.start = start;
   if (end) outgoingParams.end = end;
   if (stringifiedTimeToken) outgoingParams.string_message_token = 'true';
+  if (includeMeta) outgoingParams.include_meta = 'true';
 
   return outgoingParams;
 }
@@ -89,6 +101,14 @@ export function handleResponse(
       announce.subscription = null;
       announce.timetoken = messageEnvelope.timetoken;
       announce.message = __processMessage(modules, messageEnvelope.message);
+
+      if (messageEnvelope.actions) {
+        announce.data = messageEnvelope.actions;
+      }
+      if (messageEnvelope.meta) {
+        announce.meta = messageEnvelope.meta;
+      }
+
       response.channels[channelName].push(announce);
     });
   });

src/core/endpoints/fetch_messages.js

@@ -64,6 +64,7 @@ export function prepareParams(
     reverse,
     count = 100,
     stringifiedTimeToken = false,
+    includeMeta = false,
   } = incomingParams;
   let outgoingParams: Object = {
     include_token: 'true',
@@ -74,6 +75,7 @@ export function prepareParams(
   if (end) outgoingParams.end = end;
   if (stringifiedTimeToken) outgoingParams.string_message_token = 'true';
   if (reverse != null) outgoingParams.reverse = reverse.toString();
+  if (includeMeta) outgoingParams.include_meta = 'true';
 
   return outgoingParams;
 }
@@ -95,6 +97,10 @@ export function handleResponse(
         entry: __processMessage(modules, serverHistoryItem.message),
       };
 
+      if (serverHistoryItem.meta) {
+        item.meta = serverHistoryItem.meta;
+      }
+
       response.messages.push(item);
     });
   }