From 6911c78913ea9d6ab37c0c0eb67b7389b9108ada Mon Sep 17 00:00:00 2001
From: Anantha Krishna Bhatta <banantha@amazon.com>
Date: Mon, 9 Nov 2020 15:54:34 -0800
Subject: [PATCH] Added transport action support for sending notification API

---
 .idea/codeStyles/Project.xml                  |   9 ++
 .../notifications/NotificationPlugin.kt       |   9 +-
 .../notifications/action/PluginBaseAction.kt  | 101 +++++++++++++
 .../notifications/action/SendAction.kt        | 120 ---------------
 .../notifications/action/SendMessageAction.kt | 102 +++++++++++++
 .../notifications/channel/EmptyChannel.kt     |   7 +-
 .../channel/NotificationChannel.kt            |   4 +-
 .../channel/email/BaseEmailChannel.kt         |  19 +--
 .../channel/email/EmailMimeProvider.kt        |   2 +-
 .../notifications/channel/email/SesChannel.kt |  26 ++--
 .../channel/email/SmtpChannel.kt              |  14 +-
 .../notifications/core/NotificationMessage.kt |  71 ---------
 .../notifications/model/BaseResponse.kt       |  45 ++++++
 .../{core => model}/ChannelMessage.kt         |  34 ++++-
 .../{core => model}/ChannelMessageResponse.kt |  14 +-
 .../notifications/model/SendMessageRequest.kt | 143 ++++++++++++++++++
 .../model/SendMessageResponse.kt              | 123 +++++++++++++++
 .../resthandler/PluginRestHandler.kt          |  83 ----------
 .../RestResponseToXContentListener.kt         |  36 +++++
 ...stHandler.kt => SendMessageRestHandler.kt} |  28 ++--
 .../notifications/util/Helpers.kt             |  33 ++++
 .../NotificationsRestTestCase.kt              |   9 +-
 .../notifications/channel/SmtpChannelIT.kt    |   2 +-
 ...ndlerIT.kt => SendMessageRestHandlerIT.kt} |   2 +-
 ...ests.kt => SendMessageRestHandlerTests.kt} |   8 +-
 25 files changed, 705 insertions(+), 339 deletions(-)
 create mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/PluginBaseAction.kt
 delete mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendAction.kt
 create mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendMessageAction.kt
 delete mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/NotificationMessage.kt
 create mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/BaseResponse.kt
 rename src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/{core => model}/ChannelMessage.kt (61%)
 rename src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/{core => model}/ChannelMessageResponse.kt (85%)
 create mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageRequest.kt
 create mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageResponse.kt
 delete mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/PluginRestHandler.kt
 create mode 100644 src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/RestResponseToXContentListener.kt
 rename src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/{SendRestHandler.kt => SendMessageRestHandler.kt} (62%)
 rename src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/{SendRestHandlerIT.kt => SendMessageRestHandlerIT.kt} (92%)
 rename src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/{SendRestHandlerTests.kt => SendMessageRestHandlerTests.kt} (86%)

diff --git a/.idea/codeStyles/Project.xml b/.idea/codeStyles/Project.xml
index 006df8a3..a58b712e 100644
--- a/.idea/codeStyles/Project.xml
+++ b/.idea/codeStyles/Project.xml
@@ -8,6 +8,15 @@
       <option name="PACKAGES_TO_USE_STAR_IMPORTS">
         <value />
       </option>
+      <option name="PACKAGES_IMPORT_LAYOUT">
+        <value>
+          <package name="" alias="false" withSubpackages="true" />
+          <package name="java" alias="false" withSubpackages="true" />
+          <package name="javax" alias="false" withSubpackages="true" />
+          <package name="kotlin" alias="false" withSubpackages="true" />
+          <package name="" alias="true" withSubpackages="true" />
+        </value>
+      </option>
       <option name="NAME_COUNT_TO_USE_STAR_IMPORT" value="2147483647" />
       <option name="NAME_COUNT_TO_USE_STAR_IMPORT_FOR_MEMBERS" value="2147483647" />
     </JetCodeStyleSettings>
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationPlugin.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationPlugin.kt
index a5b22885..0fc3f8c5 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationPlugin.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationPlugin.kt
@@ -16,7 +16,8 @@
 
 package com.amazon.opendistroforelasticsearch.notifications
 
-import com.amazon.opendistroforelasticsearch.notifications.resthandler.SendRestHandler
+import com.amazon.opendistroforelasticsearch.notifications.action.SendMessageAction
+import com.amazon.opendistroforelasticsearch.notifications.resthandler.SendMessageRestHandler
 import com.amazon.opendistroforelasticsearch.notifications.settings.PluginSettings
 import com.amazon.opendistroforelasticsearch.notifications.throttle.Accountant
 import com.amazon.opendistroforelasticsearch.notifications.util.logger
@@ -97,7 +98,9 @@ internal class NotificationPlugin : ActionPlugin, Plugin() {
      */
     override fun getActions(): List<ActionPlugin.ActionHandler<out ActionRequest, out ActionResponse>> {
         log.debug("$LOG_PREFIX:getActions")
-        return listOf()
+        return listOf(
+            ActionPlugin.ActionHandler(SendMessageAction.ACTION_TYPE, SendMessageAction::class.java)
+        )
     }
 
     /**
@@ -114,7 +117,7 @@ internal class NotificationPlugin : ActionPlugin, Plugin() {
     ): List<RestHandler> {
         log.debug("$LOG_PREFIX:getRestHandlers")
         return listOf(
-            SendRestHandler()
+            SendMessageRestHandler()
         )
     }
 }
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/PluginBaseAction.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/PluginBaseAction.kt
new file mode 100644
index 00000000..31e1ff2e
--- /dev/null
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/PluginBaseAction.kt
@@ -0,0 +1,101 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.opendistroforelasticsearch.notifications.action
+
+import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
+import com.amazon.opendistroforelasticsearch.notifications.util.logger
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.launch
+import org.elasticsearch.ElasticsearchSecurityException
+import org.elasticsearch.ElasticsearchStatusException
+import org.elasticsearch.action.ActionListener
+import org.elasticsearch.action.ActionRequest
+import org.elasticsearch.action.ActionResponse
+import org.elasticsearch.action.support.ActionFilters
+import org.elasticsearch.action.support.HandledTransportAction
+import org.elasticsearch.common.io.stream.Writeable
+import org.elasticsearch.index.IndexNotFoundException
+import org.elasticsearch.index.engine.VersionConflictEngineException
+import org.elasticsearch.indices.InvalidIndexNameException
+import org.elasticsearch.rest.RestStatus
+import org.elasticsearch.tasks.Task
+import org.elasticsearch.transport.TransportService
+import java.io.IOException
+
+internal abstract class PluginBaseAction<Request : ActionRequest, Response : ActionResponse>(
+    name: String,
+    transportService: TransportService,
+    actionFilters: ActionFilters,
+    requestReader: Writeable.Reader<Request>
+) : HandledTransportAction<Request, Response>(name, transportService, actionFilters, requestReader) {
+    companion object {
+        private val log by logger(PluginBaseAction::class.java)
+        private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO)
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Suppress("TooGenericExceptionCaught")
+    override fun doExecute(
+        task: Task?,
+        request: Request,
+        listener: ActionListener<Response>
+    ) {
+        scope.launch {
+            try {
+                listener.onResponse(executeRequest(request))
+            } catch (exception: ElasticsearchStatusException) {
+                log.warn("$LOG_PREFIX:ElasticsearchStatusException:", exception)
+                listener.onFailure(exception)
+            } catch (exception: ElasticsearchSecurityException) {
+                log.warn("$LOG_PREFIX:ElasticsearchSecurityException:", exception)
+                listener.onFailure(ElasticsearchStatusException("Permissions denied: ${exception.message} - Contact administrator",
+                    RestStatus.FORBIDDEN))
+            } catch (exception: VersionConflictEngineException) {
+                log.warn("$LOG_PREFIX:VersionConflictEngineException:", exception)
+                listener.onFailure(ElasticsearchStatusException(exception.message, RestStatus.CONFLICT))
+            } catch (exception: IndexNotFoundException) {
+                log.warn("$LOG_PREFIX:IndexNotFoundException:", exception)
+                listener.onFailure(ElasticsearchStatusException(exception.message, RestStatus.NOT_FOUND))
+            } catch (exception: InvalidIndexNameException) {
+                log.warn("$LOG_PREFIX:InvalidIndexNameException:", exception)
+                listener.onFailure(ElasticsearchStatusException(exception.message, RestStatus.BAD_REQUEST))
+            } catch (exception: IllegalArgumentException) {
+                log.warn("$LOG_PREFIX:IllegalArgumentException:", exception)
+                listener.onFailure(ElasticsearchStatusException(exception.message, RestStatus.BAD_REQUEST))
+            } catch (exception: IllegalStateException) {
+                log.warn("$LOG_PREFIX:IllegalStateException:", exception)
+                listener.onFailure(ElasticsearchStatusException(exception.message, RestStatus.SERVICE_UNAVAILABLE))
+            } catch (exception: IOException) {
+                log.error("$LOG_PREFIX:Uncaught IOException:", exception)
+                listener.onFailure(ElasticsearchStatusException(exception.message, RestStatus.FAILED_DEPENDENCY))
+            } catch (exception: Exception) {
+                log.error("$LOG_PREFIX:Uncaught Exception:", exception)
+                listener.onFailure(ElasticsearchStatusException(exception.message, RestStatus.INTERNAL_SERVER_ERROR))
+            }
+        }
+    }
+
+    /**
+     * Execute the transport request
+     * @param request the request to execute
+     * @return the response to return.
+     */
+    abstract fun executeRequest(request: Request): Response
+}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendAction.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendAction.kt
deleted file mode 100644
index afde8f34..00000000
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendAction.kt
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License").
- * You may not use this file except in compliance with the License.
- * A copy of the License is located at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * or in the "license" file accompanying this file. This file is distributed
- * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
- * express or implied. See the License for the specific language governing
- * permissions and limitations under the License.
- *
- */
-
-package com.amazon.opendistroforelasticsearch.notifications.action
-
-import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
-import com.amazon.opendistroforelasticsearch.notifications.channel.ChannelFactory
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessageResponse
-import com.amazon.opendistroforelasticsearch.notifications.core.NotificationMessage
-import com.amazon.opendistroforelasticsearch.notifications.throttle.Accountant
-import com.amazon.opendistroforelasticsearch.notifications.throttle.Counters
-import com.amazon.opendistroforelasticsearch.notifications.util.logger
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.runBlocking
-import org.elasticsearch.client.node.NodeClient
-import org.elasticsearch.common.xcontent.XContentType
-import org.elasticsearch.rest.BytesRestResponse
-import org.elasticsearch.rest.RestChannel
-import org.elasticsearch.rest.RestRequest
-import org.elasticsearch.rest.RestStatus
-
-/**
- * Send action for send notification request.
- */
-internal class SendAction(
-    private val request: RestRequest,
-    private val client: NodeClient,
-    private val restChannel: RestChannel
-) {
-
-    internal companion object {
-        private val log by logger(SendAction::class.java)
-    }
-
-    /**
-     * Send notification for the given [request] on the provided [restChannel].
-     */
-    fun send() {
-        log.debug("$LOG_PREFIX:send")
-        val contentParser = request.contentParser()
-        contentParser.nextToken()
-        val message = NotificationMessage.parse(contentParser)
-        val response = restChannel.newBuilder(XContentType.JSON, false).startObject()
-            .field("refTag", message.refTag)
-        var restStatus = RestStatus.OK // Default to success
-        if (isMessageQuotaAvailable(message)) {
-            response.startArray("recipients")
-            val statusList: List<Pair<String, ChannelMessageResponse>> = sendMessagesInParallel(message)
-            // Get all the response in sequence
-            statusList.forEach {
-                val statusCode = it.second.statusCode
-                val statusText = it.second.statusText
-                if (statusCode != RestStatus.OK && statusCode != restStatus) {
-                    // if any of the value != success then return corresponding status or 207
-                    restStatus = RestStatus.MULTI_STATUS
-                }
-                response.startObject()
-                    .field("recipient", it.first)
-                    .field("statusCode", statusCode.status)
-                    .field("statusText", statusText)
-                    .endObject()
-                log.info("$LOG_PREFIX:${message.refTag}:statusCode=$statusCode, statusText=$statusText")
-            }
-            response.endArray()
-        } else {
-            restStatus = RestStatus.TOO_MANY_REQUESTS
-            val statusText = "Message Sending quota not available"
-            response.field("statusCode", restStatus)
-                .field("statusText", statusText)
-            log.info("$LOG_PREFIX:${message.refTag}:statusCode=$restStatus, statusText=$statusText")
-        }
-        response.endObject()
-        restChannel.sendResponse(BytesRestResponse(restStatus, response))
-    }
-
-    private fun sendMessagesInParallel(message: NotificationMessage): List<Pair<String, ChannelMessageResponse>> {
-        val counters = Counters()
-        counters.requestCount.incrementAndGet()
-        val statusList: List<Pair<String, ChannelMessageResponse>>
-        // Fire all the message sending in parallel
-        runBlocking {
-            val statusDeferredList = message.recipients.map {
-                async(Dispatchers.IO) { sendMessageToChannel(it, message, counters) }
-            }
-            statusList = statusDeferredList.awaitAll()
-        }
-        // After all operation are executed, update the counters
-        Accountant.incrementCounters(counters)
-        return statusList
-    }
-
-    private fun sendMessageToChannel(recipient: String, message: NotificationMessage, counters: Counters): Pair<String, ChannelMessageResponse> {
-        val channel = ChannelFactory.getNotificationChannel(recipient)
-        val status = channel.sendMessage(message.refTag, recipient, message.channelMessage, counters)
-        return Pair(recipient, status)
-    }
-
-    private fun isMessageQuotaAvailable(message: NotificationMessage): Boolean {
-        val counters = Counters()
-        message.recipients.forEach {
-            ChannelFactory.getNotificationChannel(it).updateCounter(message.refTag, it, message.channelMessage, counters)
-        }
-        return Accountant.isMessageQuotaAvailable(counters)
-    }
-}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendMessageAction.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendMessageAction.kt
new file mode 100644
index 00000000..e0cab9d0
--- /dev/null
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/action/SendMessageAction.kt
@@ -0,0 +1,102 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.opendistroforelasticsearch.notifications.action
+
+import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
+import com.amazon.opendistroforelasticsearch.notifications.channel.ChannelFactory
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessageResponse
+import com.amazon.opendistroforelasticsearch.notifications.model.SendMessageRequest
+import com.amazon.opendistroforelasticsearch.notifications.model.SendMessageResponse
+import com.amazon.opendistroforelasticsearch.notifications.throttle.Accountant
+import com.amazon.opendistroforelasticsearch.notifications.throttle.Counters
+import com.amazon.opendistroforelasticsearch.notifications.util.logger
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.runBlocking
+import org.elasticsearch.ElasticsearchStatusException
+import org.elasticsearch.action.ActionType
+import org.elasticsearch.action.support.ActionFilters
+import org.elasticsearch.client.Client
+import org.elasticsearch.common.inject.Inject
+import org.elasticsearch.common.xcontent.NamedXContentRegistry
+import org.elasticsearch.rest.RestStatus
+import org.elasticsearch.transport.TransportService
+
+/**
+ * Send message action for send notification request.
+ */
+internal class SendMessageAction @Inject constructor(
+    transportService: TransportService,
+    val client: Client,
+    actionFilters: ActionFilters,
+    val xContentRegistry: NamedXContentRegistry
+) : PluginBaseAction<SendMessageRequest, SendMessageResponse>(NAME,
+    transportService,
+    actionFilters,
+    ::SendMessageRequest) {
+    companion object {
+        private const val NAME = "cluster:admin/opendistro/notifications/send"
+        internal val ACTION_TYPE = ActionType(NAME, ::SendMessageResponse)
+        private val log by logger(SendMessageAction::class.java)
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    override fun executeRequest(request: SendMessageRequest): SendMessageResponse {
+        log.debug("$LOG_PREFIX:send")
+        if (!isMessageQuotaAvailable(request)) {
+            log.info("$LOG_PREFIX:${request.refTag}:Message Sending quota not available")
+            throw ElasticsearchStatusException("Message Sending quota not available", RestStatus.TOO_MANY_REQUESTS)
+        }
+        val statusList: List<ChannelMessageResponse> = sendMessagesInParallel(request)
+        statusList.forEach {
+            log.info("$LOG_PREFIX:${request.refTag}:statusCode=${it.statusCode}, statusText=${it.statusText}")
+        }
+        return SendMessageResponse(request.refTag, statusList)
+    }
+
+    private fun sendMessagesInParallel(sendMessageRequest: SendMessageRequest): List<ChannelMessageResponse> {
+        val counters = Counters()
+        counters.requestCount.incrementAndGet()
+        val statusList: List<ChannelMessageResponse>
+        // Fire all the message sending in parallel
+        runBlocking {
+            val statusDeferredList = sendMessageRequest.recipients.map {
+                async(Dispatchers.IO) { sendMessageToChannel(it, sendMessageRequest, counters) }
+            }
+            statusList = statusDeferredList.awaitAll()
+        }
+        // After all operation are executed, update the counters
+        Accountant.incrementCounters(counters)
+        return statusList
+    }
+
+    private fun sendMessageToChannel(recipient: String, sendMessageRequest: SendMessageRequest, counters: Counters): ChannelMessageResponse {
+        val channel = ChannelFactory.getNotificationChannel(recipient)
+        return channel.sendMessage(sendMessageRequest.refTag, recipient, sendMessageRequest.channelMessage, counters)
+    }
+
+    private fun isMessageQuotaAvailable(sendMessageRequest: SendMessageRequest): Boolean {
+        val counters = Counters()
+        sendMessageRequest.recipients.forEach {
+            ChannelFactory.getNotificationChannel(it).updateCounter(sendMessageRequest.refTag, it, sendMessageRequest.channelMessage, counters)
+        }
+        return Accountant.isMessageQuotaAvailable(counters)
+    }
+}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/EmptyChannel.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/EmptyChannel.kt
index 9138ac37..7fdc56ab 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/EmptyChannel.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/EmptyChannel.kt
@@ -16,8 +16,8 @@
 
 package com.amazon.opendistroforelasticsearch.notifications.channel
 
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessage
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessageResponse
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessage
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessageResponse
 import com.amazon.opendistroforelasticsearch.notifications.throttle.Counters
 import org.elasticsearch.rest.RestStatus
 
@@ -29,7 +29,8 @@ internal object EmptyChannel : NotificationChannel {
      * {@inheritDoc}
      */
     override fun sendMessage(refTag: String, recipient: String, channelMessage: ChannelMessage, counter: Counters): ChannelMessageResponse {
-        return ChannelMessageResponse(RestStatus.UNPROCESSABLE_ENTITY,
+        return ChannelMessageResponse(recipient,
+            RestStatus.UNPROCESSABLE_ENTITY,
             "No Configured Channel for recipient type:${recipient.substringBefore(':', "empty")}")
     }
 
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/NotificationChannel.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/NotificationChannel.kt
index 0741d7af..a316d98d 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/NotificationChannel.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/NotificationChannel.kt
@@ -16,8 +16,8 @@
 
 package com.amazon.opendistroforelasticsearch.notifications.channel
 
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessage
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessageResponse
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessage
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessageResponse
 import com.amazon.opendistroforelasticsearch.notifications.throttle.Counters
 
 /**
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/BaseEmailChannel.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/BaseEmailChannel.kt
index c9c60f3f..01ca8929 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/BaseEmailChannel.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/BaseEmailChannel.kt
@@ -17,8 +17,8 @@
 package com.amazon.opendistroforelasticsearch.notifications.channel.email
 
 import com.amazon.opendistroforelasticsearch.notifications.channel.NotificationChannel
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessage
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessageResponse
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessage
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessageResponse
 import com.amazon.opendistroforelasticsearch.notifications.settings.PluginSettings
 import com.amazon.opendistroforelasticsearch.notifications.throttle.Counters
 import org.elasticsearch.rest.RestStatus
@@ -67,22 +67,22 @@ internal abstract class BaseEmailChannel : NotificationChannel {
     private fun sendEmail(refTag: String, recipient: String, channelMessage: ChannelMessage): ChannelMessageResponse {
         val fromAddress = PluginSettings.emailFromAddress
         if (PluginSettings.UNCONFIGURED_EMAIL_ADDRESS == fromAddress) {
-            return ChannelMessageResponse(RestStatus.NOT_IMPLEMENTED, "Email from: address not configured")
+            return ChannelMessageResponse(recipient, RestStatus.NOT_IMPLEMENTED, "Email from: address not configured")
         }
         if (isMessageSizeOverLimit(channelMessage)) {
-            return ChannelMessageResponse(RestStatus.REQUEST_ENTITY_TOO_LARGE, "Email size larger than ${PluginSettings.emailSizeLimit}")
+            return ChannelMessageResponse(recipient, RestStatus.REQUEST_ENTITY_TOO_LARGE, "Email size larger than ${PluginSettings.emailSizeLimit}")
         }
         val mimeMessage: MimeMessage
         return try {
             val session = prepareSession(refTag, recipient, channelMessage)
             mimeMessage = EmailMimeProvider.prepareMimeMessage(session, fromAddress, recipient, channelMessage)
-            sendMimeMessage(refTag, mimeMessage)
+            sendMimeMessage(refTag, recipient, mimeMessage)
         } catch (addressException: AddressException) {
-            ChannelMessageResponse(RestStatus.BAD_REQUEST, "recipient parsing failed with status:${addressException.message}")
+            ChannelMessageResponse(recipient, RestStatus.BAD_REQUEST, "recipient parsing failed with status:${addressException.message}")
         } catch (messagingException: MessagingException) {
-            ChannelMessageResponse(RestStatus.FAILED_DEPENDENCY, "Email message creation failed with status:${messagingException.message}")
+            ChannelMessageResponse(recipient, RestStatus.FAILED_DEPENDENCY, "Email message creation failed with status:${messagingException.message}")
         } catch (ioException: IOException) {
-            ChannelMessageResponse(RestStatus.FAILED_DEPENDENCY, "Email message creation failed with status:${ioException.message}")
+            ChannelMessageResponse(recipient, RestStatus.FAILED_DEPENDENCY, "Email message creation failed with status:${ioException.message}")
         }
     }
 
@@ -114,8 +114,9 @@ internal abstract class BaseEmailChannel : NotificationChannel {
     /**
      * Sending Email mime message to server.
      * @param refTag ref tag for logging purpose
+     * @param recipient email recipient to send mail to
      * @param mimeMessage mime message to send to Email server
      * @return Channel message response
      */
-    protected abstract fun sendMimeMessage(refTag: String, mimeMessage: MimeMessage): ChannelMessageResponse
+    protected abstract fun sendMimeMessage(refTag: String, recipient: String, mimeMessage: MimeMessage): ChannelMessageResponse
 }
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/EmailMimeProvider.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/EmailMimeProvider.kt
index 2db39aff..fe850f3a 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/EmailMimeProvider.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/EmailMimeProvider.kt
@@ -16,7 +16,7 @@
 
 package com.amazon.opendistroforelasticsearch.notifications.channel.email
 
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessage
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessage
 import java.util.Base64
 import javax.activation.DataHandler
 import javax.mail.Message
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SesChannel.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SesChannel.kt
index 015099c5..ba263ee6 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SesChannel.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SesChannel.kt
@@ -17,8 +17,8 @@
 package com.amazon.opendistroforelasticsearch.notifications.channel.email
 
 import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessage
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessageResponse
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessage
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessageResponse
 import com.amazon.opendistroforelasticsearch.notifications.security.SecurityAccess
 import com.amazon.opendistroforelasticsearch.notifications.settings.PluginSettings
 import com.amazon.opendistroforelasticsearch.notifications.util.logger
@@ -59,7 +59,7 @@ internal object SesChannel : BaseEmailChannel() {
     /**
      * {@inheritDoc}
      */
-    override fun sendMimeMessage(refTag: String, mimeMessage: MimeMessage): ChannelMessageResponse {
+    override fun sendMimeMessage(refTag: String, recipient: String, mimeMessage: MimeMessage): ChannelMessageResponse {
         return try {
             log.debug("$LOG_PREFIX:Sending Email-SES:$refTag")
             val region = Region.of(PluginSettings.sesAwsRegion)
@@ -79,24 +79,26 @@ internal object SesChannel : BaseEmailChannel() {
                     .build()
                 val response = SecurityAccess.doPrivileged { client.sendRawEmail(rawEmailRequest) }
                 log.info("$LOG_PREFIX:Email-SES:$refTag status:$response")
-                ChannelMessageResponse(RestStatus.OK, "Success")
+                ChannelMessageResponse(recipient, RestStatus.OK, "Success")
             } else {
-                ChannelMessageResponse(RestStatus.REQUEST_ENTITY_TOO_LARGE, "Email size($emailSize) larger than ${PluginSettings.emailSizeLimit}")
+                ChannelMessageResponse(recipient,
+                    RestStatus.REQUEST_ENTITY_TOO_LARGE,
+                    "Email size($emailSize) larger than ${PluginSettings.emailSizeLimit}")
             }
         } catch (exception: MessageRejectedException) {
-            ChannelMessageResponse(RestStatus.SERVICE_UNAVAILABLE, getSesExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.SERVICE_UNAVAILABLE, getSesExceptionText(exception))
         } catch (exception: MailFromDomainNotVerifiedException) {
-            ChannelMessageResponse(RestStatus.FORBIDDEN, getSesExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.FORBIDDEN, getSesExceptionText(exception))
         } catch (exception: ConfigurationSetDoesNotExistException) {
-            ChannelMessageResponse(RestStatus.NOT_IMPLEMENTED, getSesExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.NOT_IMPLEMENTED, getSesExceptionText(exception))
         } catch (exception: ConfigurationSetSendingPausedException) {
-            ChannelMessageResponse(RestStatus.SERVICE_UNAVAILABLE, getSesExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.SERVICE_UNAVAILABLE, getSesExceptionText(exception))
         } catch (exception: AccountSendingPausedException) {
-            ChannelMessageResponse(RestStatus.INSUFFICIENT_STORAGE, getSesExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.INSUFFICIENT_STORAGE, getSesExceptionText(exception))
         } catch (exception: SesException) {
-            ChannelMessageResponse(RestStatus.FAILED_DEPENDENCY, getSesExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.FAILED_DEPENDENCY, getSesExceptionText(exception))
         } catch (exception: SdkException) {
-            ChannelMessageResponse(RestStatus.FAILED_DEPENDENCY, getSdkExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.FAILED_DEPENDENCY, getSdkExceptionText(exception))
         }
     }
 
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SmtpChannel.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SmtpChannel.kt
index a91246c3..eca47e2e 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SmtpChannel.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/email/SmtpChannel.kt
@@ -17,8 +17,8 @@
 package com.amazon.opendistroforelasticsearch.notifications.channel.email
 
 import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessage
-import com.amazon.opendistroforelasticsearch.notifications.core.ChannelMessageResponse
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessage
+import com.amazon.opendistroforelasticsearch.notifications.model.ChannelMessageResponse
 import com.amazon.opendistroforelasticsearch.notifications.security.SecurityAccess
 import com.amazon.opendistroforelasticsearch.notifications.settings.PluginSettings
 import com.amazon.opendistroforelasticsearch.notifications.util.logger
@@ -55,18 +55,18 @@ internal object SmtpChannel : BaseEmailChannel() {
     /**
      * {@inheritDoc}
      */
-    override fun sendMimeMessage(refTag: String, mimeMessage: MimeMessage): ChannelMessageResponse {
+    override fun sendMimeMessage(refTag: String, recipient: String, mimeMessage: MimeMessage): ChannelMessageResponse {
         return try {
             log.debug("$LOG_PREFIX:Sending Email-SMTP:$refTag")
             SecurityAccess.doPrivileged { Transport.send(mimeMessage) }
             log.info("$LOG_PREFIX:Email-SMTP:$refTag sent")
-            ChannelMessageResponse(RestStatus.OK, "Success")
+            ChannelMessageResponse(recipient, RestStatus.OK, "Success")
         } catch (exception: SendFailedException) {
-            ChannelMessageResponse(RestStatus.BAD_GATEWAY, getMessagingExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.BAD_GATEWAY, getMessagingExceptionText(exception))
         } catch (exception: MailConnectException) {
-            ChannelMessageResponse(RestStatus.SERVICE_UNAVAILABLE, getMessagingExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.SERVICE_UNAVAILABLE, getMessagingExceptionText(exception))
         } catch (exception: MessagingException) {
-            ChannelMessageResponse(RestStatus.FAILED_DEPENDENCY, getMessagingExceptionText(exception))
+            ChannelMessageResponse(recipient, RestStatus.FAILED_DEPENDENCY, getMessagingExceptionText(exception))
         }
     }
 
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/NotificationMessage.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/NotificationMessage.kt
deleted file mode 100644
index ed0b9a78..00000000
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/NotificationMessage.kt
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License").
- * You may not use this file except in compliance with the License.
- * A copy of the License is located at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * or in the "license" file accompanying this file. This file is distributed
- * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
- * express or implied. See the License for the specific language governing
- * permissions and limitations under the License.
- *
- */
-
-package com.amazon.opendistroforelasticsearch.notifications.core
-
-import com.amazon.opendistroforelasticsearch.notifications.util.stringList
-import org.elasticsearch.common.xcontent.XContentParser
-import org.elasticsearch.common.xcontent.XContentParserUtils
-
-/**
- * Data class for storing the notification message.
- */
-internal data class NotificationMessage(
-    val refTag: String,
-    val recipients: List<String>,
-    val channelMessage: ChannelMessage
-) {
-    internal companion object {
-        /**
-         * Parse the data from parser and create Attachment object
-         * @param parser data referenced at parser
-         * @return created Attachment object
-         */
-        fun parse(parser: XContentParser): NotificationMessage {
-            var refTag: String? = null
-            var title: String? = null
-            var textDescription: String? = null
-            var htmlDescription: String? = null
-            var attachment: ChannelMessage.Attachment? = null
-            var recipients: List<String> = listOf()
-            XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser::getTokenLocation)
-            while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
-                val fieldName = parser.currentName()
-                parser.nextToken()
-                when (fieldName) {
-                    "refTag" -> refTag = parser.text()
-                    "recipients" -> recipients = parser.stringList()
-                    "title" -> title = parser.text()
-                    "textDescription" -> textDescription = parser.text()
-                    "htmlDescription" -> htmlDescription = parser.text()
-                    "attachment" -> attachment = ChannelMessage.Attachment.parse(parser)
-                    else -> {
-                        parser.skipChildren()
-                    }
-                }
-            }
-            refTag = refTag ?: "noRef"
-            if (recipients.isEmpty()) {
-                throw IllegalArgumentException("Empty recipient list")
-            }
-            title ?: throw IllegalArgumentException("Title field not present")
-            textDescription ?: throw IllegalArgumentException("textDescription not present")
-            return NotificationMessage(refTag,
-                recipients.toList(),
-                ChannelMessage(title, textDescription, htmlDescription, attachment))
-        }
-    }
-}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/BaseResponse.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/BaseResponse.kt
new file mode 100644
index 00000000..19a0c2d2
--- /dev/null
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/BaseResponse.kt
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.opendistroforelasticsearch.notifications.model
+
+import org.elasticsearch.action.ActionResponse
+import org.elasticsearch.common.xcontent.ToXContent
+import org.elasticsearch.common.xcontent.ToXContentObject
+import org.elasticsearch.common.xcontent.XContentBuilder
+import org.elasticsearch.common.xcontent.XContentFactory
+import org.elasticsearch.rest.RestStatus
+
+/**
+ * Base response which give REST status.
+ */
+internal abstract class BaseResponse : ActionResponse(), ToXContentObject {
+
+    /**
+     * {@inheritDoc}
+     */
+    fun toXContent(): XContentBuilder {
+        return toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)
+    }
+
+    /**
+     * get rest status for the response. Useful override for multi-status response.
+     * @return RestStatus for the response
+     */
+    open fun getStatus(): RestStatus {
+        return RestStatus.OK
+    }
+}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/ChannelMessage.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/ChannelMessage.kt
similarity index 61%
rename from src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/ChannelMessage.kt
rename to src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/ChannelMessage.kt
index 7bb9d34e..43ea557c 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/ChannelMessage.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/ChannelMessage.kt
@@ -14,8 +14,14 @@
  *
  */
 
-package com.amazon.opendistroforelasticsearch.notifications.core
+package com.amazon.opendistroforelasticsearch.notifications.model
 
+import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
+import com.amazon.opendistroforelasticsearch.notifications.util.fieldIfNotNull
+import com.amazon.opendistroforelasticsearch.notifications.util.logger
+import org.elasticsearch.common.xcontent.ToXContent
+import org.elasticsearch.common.xcontent.ToXContentObject
+import org.elasticsearch.common.xcontent.XContentBuilder
 import org.elasticsearch.common.xcontent.XContentParser
 import org.elasticsearch.common.xcontent.XContentParserUtils
 
@@ -36,8 +42,14 @@ internal data class ChannelMessage(
         val fileEncoding: String,
         val fileData: String,
         val fileContentType: String?
-    ) {
+    ) : ToXContentObject {
         internal companion object {
+            private val log by logger(Attachment::class.java)
+            private const val FILE_NAME_FIELD = "fileName"
+            private const val FILE_ENCODING_FIELD = "fileEncoding"
+            private const val FILE_DATA_FIELD = "fileData"
+            private const val FILE_CONTENT_TYPE_FIELD = "fileContentType"
+
             /**
              * Parse the data from parser and create Attachment object
              * @param parser data referenced at parser
@@ -53,12 +65,13 @@ internal data class ChannelMessage(
                     val dataType = parser.currentName()
                     parser.nextToken()
                     when (dataType) {
-                        "fileName" -> fileName = parser.text()
-                        "fileEncoding" -> fileEncoding = parser.text()
-                        "fileData" -> fileData = parser.text()
-                        "fileContentType" -> fileContentType = parser.text()
+                        FILE_NAME_FIELD -> fileName = parser.text()
+                        FILE_ENCODING_FIELD -> fileEncoding = parser.text()
+                        FILE_DATA_FIELD -> fileData = parser.text()
+                        FILE_CONTENT_TYPE_FIELD -> fileContentType = parser.text()
                         else -> {
                             parser.skipChildren()
+                            log.info("$LOG_PREFIX:Skipping Unknown field $dataType")
                         }
                     }
                 }
@@ -68,5 +81,14 @@ internal data class ChannelMessage(
                 return Attachment(fileName, fileEncoding, fileData, fileContentType)
             }
         }
+
+        override fun toXContent(builder: XContentBuilder?, params: ToXContent.Params?): XContentBuilder {
+            return builder!!.startObject()
+                .field(FILE_NAME_FIELD, fileName)
+                .field(FILE_ENCODING_FIELD, fileEncoding)
+                .field(FILE_DATA_FIELD, fileData)
+                .fieldIfNotNull(FILE_CONTENT_TYPE_FIELD, fileContentType)
+                .endObject()
+        }
     }
 }
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/ChannelMessageResponse.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/ChannelMessageResponse.kt
similarity index 85%
rename from src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/ChannelMessageResponse.kt
rename to src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/ChannelMessageResponse.kt
index 77e0f388..8000c159 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/core/ChannelMessageResponse.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/ChannelMessageResponse.kt
@@ -14,7 +14,7 @@
  *
  */
 
-package com.amazon.opendistroforelasticsearch.notifications.core
+package com.amazon.opendistroforelasticsearch.notifications.model
 
 import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
 import com.amazon.opendistroforelasticsearch.notifications.util.logger
@@ -30,13 +30,15 @@ import org.elasticsearch.rest.RestStatus
  * Data class for storing channel message response per recipient.
  */
 internal data class ChannelMessageResponse(
+    val recipient: String,
     val statusCode: RestStatus,
     val statusText: String
 ) : ToXContentObject {
     internal companion object {
         private val log by logger(ChannelMessageResponse::class.java)
-        const val STATUS_CODE_TAG = "statusCode"
-        const val STATUS_TEXT_TAG = "statusText"
+        private const val RECIPIENT_TAG = "recipient"
+        private const val STATUS_CODE_TAG = "statusCode"
+        private const val STATUS_TEXT_TAG = "statusText"
 
         /**
          * Parse the data from parser and create ChannelMessageResponse object
@@ -44,6 +46,7 @@ internal data class ChannelMessageResponse(
          * @return created ChannelMessageResponse object
          */
         fun parse(parser: XContentParser): ChannelMessageResponse {
+            var recipient: String? = null
             var statusCode: RestStatus? = null
             var statusText: String? = null
             XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser::getTokenLocation)
@@ -51,6 +54,7 @@ internal data class ChannelMessageResponse(
                 val fieldName = parser.currentName()
                 parser.nextToken()
                 when (fieldName) {
+                    RECIPIENT_TAG -> recipient = parser.text()
                     STATUS_CODE_TAG -> statusCode = RestStatus.fromCode(parser.intValue())
                     STATUS_TEXT_TAG -> statusText = parser.text()
                     else -> {
@@ -59,9 +63,10 @@ internal data class ChannelMessageResponse(
                     }
                 }
             }
+            recipient ?: throw IllegalArgumentException("$RECIPIENT_TAG field absent")
             statusCode ?: throw IllegalArgumentException("$STATUS_CODE_TAG field absent")
             statusText ?: throw IllegalArgumentException("$STATUS_TEXT_TAG field absent")
-            return ChannelMessageResponse(statusCode, statusText)
+            return ChannelMessageResponse(recipient, statusCode, statusText)
         }
     }
 
@@ -79,6 +84,7 @@ internal data class ChannelMessageResponse(
     override fun toXContent(builder: XContentBuilder?, params: ToXContent.Params?): XContentBuilder {
         builder!!
         builder.startObject()
+            .field(RECIPIENT_TAG, recipient)
             .field(STATUS_CODE_TAG, statusCode.status)
             .field(STATUS_TEXT_TAG, statusText)
             .endObject()
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageRequest.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageRequest.kt
new file mode 100644
index 00000000..e82b2669
--- /dev/null
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageRequest.kt
@@ -0,0 +1,143 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.opendistroforelasticsearch.notifications.model
+
+import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
+import com.amazon.opendistroforelasticsearch.notifications.util.createJsonParser
+import com.amazon.opendistroforelasticsearch.notifications.util.fieldIfNotNull
+import com.amazon.opendistroforelasticsearch.notifications.util.logger
+import com.amazon.opendistroforelasticsearch.notifications.util.objectIfNotNull
+import com.amazon.opendistroforelasticsearch.notifications.util.stringList
+import org.elasticsearch.action.ActionRequest
+import org.elasticsearch.action.ActionRequestValidationException
+import org.elasticsearch.common.io.stream.StreamInput
+import org.elasticsearch.common.io.stream.StreamOutput
+import org.elasticsearch.common.xcontent.ToXContent
+import org.elasticsearch.common.xcontent.ToXContentObject
+import org.elasticsearch.common.xcontent.XContentBuilder
+import org.elasticsearch.common.xcontent.XContentFactory
+import org.elasticsearch.common.xcontent.XContentParser
+import org.elasticsearch.common.xcontent.XContentParserUtils
+import java.io.IOException
+
+/**
+ * Data class for storing the send message request.
+ */
+internal class SendMessageRequest : ActionRequest, ToXContentObject {
+    val refTag: String
+    val recipients: List<String>
+    val channelMessage: ChannelMessage
+
+    companion object {
+        private val log by logger(SendMessageRequest::class.java)
+        private const val REF_TAG_FIELD = "refTag"
+        private const val RECIPIENTS_FIELD = "recipients"
+        private const val TITLE_FIELD = "title"
+        private const val TEXT_DESCRIPTION_FIELD = "textDescription"
+        private const val HTML_DESCRIPTION_FIELD = "htmlDescription"
+        private const val ATTACHMENT_FIELD = "attachment"
+    }
+
+    constructor(
+        refTag: String,
+        recipients: List<String>,
+        channelMessage: ChannelMessage
+    ) : super() {
+        this.refTag = refTag
+        this.recipients = recipients
+        this.channelMessage = channelMessage
+    }
+
+    @Throws(IOException::class)
+    constructor(input: StreamInput) : this(input.createJsonParser())
+
+    /**
+     * Parse the data from parser and create object
+     * @param parser data referenced at parser
+     */
+    constructor(parser: XContentParser) : super() {
+        var refTag: String? = null
+        var title: String? = null
+        var textDescription: String? = null
+        var htmlDescription: String? = null
+        var attachment: ChannelMessage.Attachment? = null
+        var recipients: List<String> = listOf()
+        XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser::getTokenLocation)
+        while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
+            val fieldName = parser.currentName()
+            parser.nextToken()
+            when (fieldName) {
+                REF_TAG_FIELD -> refTag = parser.text()
+                RECIPIENTS_FIELD -> recipients = parser.stringList()
+                TITLE_FIELD -> title = parser.text()
+                TEXT_DESCRIPTION_FIELD -> textDescription = parser.text()
+                HTML_DESCRIPTION_FIELD -> htmlDescription = parser.text()
+                ATTACHMENT_FIELD -> attachment = ChannelMessage.Attachment.parse(parser)
+                else -> {
+                    parser.skipChildren()
+                    log.info("$LOG_PREFIX:Skipping Unknown field $fieldName")
+                }
+            }
+        }
+        refTag = refTag ?: "noRef"
+        if (recipients.isEmpty()) {
+            throw IllegalArgumentException("Empty recipient list")
+        }
+        title ?: throw IllegalArgumentException("$TITLE_FIELD field not present")
+        textDescription ?: throw IllegalArgumentException("$TEXT_DESCRIPTION_FIELD not present")
+        this.refTag = refTag
+        this.recipients = recipients
+        this.channelMessage = ChannelMessage(title, textDescription, htmlDescription, attachment)
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Throws(IOException::class)
+    override fun writeTo(output: StreamOutput) {
+        toXContent(XContentFactory.jsonBuilder(output), ToXContent.EMPTY_PARAMS)
+    }
+
+    /**
+     * create XContentBuilder from this object using [XContentFactory.jsonBuilder()]
+     * @return created XContentBuilder object
+     */
+    fun toXContent(): XContentBuilder? {
+        return toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    override fun toXContent(builder: XContentBuilder?, params: ToXContent.Params?): XContentBuilder {
+        return builder!!.startObject()
+            .field(REF_TAG_FIELD, refTag)
+            .field(RECIPIENTS_FIELD, recipients)
+            .field(TITLE_FIELD, channelMessage.title)
+            .field(TEXT_DESCRIPTION_FIELD, channelMessage.textDescription)
+            .fieldIfNotNull(HTML_DESCRIPTION_FIELD, channelMessage.htmlDescription)
+            .objectIfNotNull(ATTACHMENT_FIELD, channelMessage.attachment)
+            .endObject()
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    override fun validate(): ActionRequestValidationException? {
+        return null
+    }
+}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageResponse.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageResponse.kt
new file mode 100644
index 00000000..808db8eb
--- /dev/null
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/model/SendMessageResponse.kt
@@ -0,0 +1,123 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.opendistroforelasticsearch.notifications.model
+
+import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
+import com.amazon.opendistroforelasticsearch.notifications.util.createJsonParser
+import com.amazon.opendistroforelasticsearch.notifications.util.logger
+import org.elasticsearch.common.io.stream.StreamInput
+import org.elasticsearch.common.io.stream.StreamOutput
+import org.elasticsearch.common.xcontent.ToXContent
+import org.elasticsearch.common.xcontent.XContentBuilder
+import org.elasticsearch.common.xcontent.XContentFactory
+import org.elasticsearch.common.xcontent.XContentParser
+import org.elasticsearch.common.xcontent.XContentParser.Token
+import org.elasticsearch.common.xcontent.XContentParserUtils
+import org.elasticsearch.rest.RestStatus
+import java.io.IOException
+
+/**
+ * data class for storing send message response.
+ */
+internal class SendMessageResponse : BaseResponse {
+    val refTag: String
+    val channelMessageResponseList: List<ChannelMessageResponse>
+
+    companion object {
+        private val log by logger(SendMessageResponse::class.java)
+        private const val REF_TAG_FIELD = "refTag"
+        private const val RECIPIENTS_TAG_FIELD = "recipients"
+
+        /**
+         * Parse the recipient response list from parser
+         * @param parser data referenced at parser
+         * @return created list of ReportInstance
+         */
+        private fun parseRecipientResponseList(parser: XContentParser): List<ChannelMessageResponse> {
+            val retList: MutableList<ChannelMessageResponse> = mutableListOf()
+            XContentParserUtils.ensureExpectedToken(Token.START_ARRAY, parser.currentToken(), parser::getTokenLocation)
+            while (parser.nextToken() != Token.END_ARRAY) {
+                retList.add(ChannelMessageResponse.parse(parser))
+            }
+            return retList
+        }
+    }
+
+    constructor(refTag: String, channelMessageResponseList: List<ChannelMessageResponse>) : super() {
+        this.refTag = refTag
+        this.channelMessageResponseList = channelMessageResponseList
+    }
+
+    @Throws(IOException::class)
+    constructor(input: StreamInput) : this(input.createJsonParser())
+
+    /**
+     * Parse the data from parser and create object
+     * @param parser data referenced at parser
+     */
+    constructor(parser: XContentParser) : super() {
+        var refTag: String? = null
+        var channelMessageResponseList: List<ChannelMessageResponse>? = null
+        XContentParserUtils.ensureExpectedToken(Token.START_OBJECT, parser.currentToken(), parser::getTokenLocation)
+        while (Token.END_OBJECT != parser.nextToken()) {
+            val fieldName = parser.currentName()
+            parser.nextToken()
+            when (fieldName) {
+                REF_TAG_FIELD -> refTag = parser.text()
+                RECIPIENTS_TAG_FIELD -> channelMessageResponseList = parseRecipientResponseList(parser)
+                else -> {
+                    parser.skipChildren()
+                    log.info("$LOG_PREFIX:Skipping Unknown field $fieldName")
+                }
+            }
+        }
+        refTag = refTag ?: "noRef"
+        channelMessageResponseList ?: throw IllegalArgumentException("$RECIPIENTS_TAG_FIELD field absent")
+        this.refTag = refTag
+        this.channelMessageResponseList = channelMessageResponseList
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Throws(IOException::class)
+    override fun writeTo(output: StreamOutput) {
+        toXContent(XContentFactory.jsonBuilder(output), ToXContent.EMPTY_PARAMS)
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    override fun getStatus(): RestStatus {
+        val resultList = channelMessageResponseList.map { it.statusCode }.distinct()
+        return if (resultList.size == 1) {
+            resultList[0]
+        } else {
+            RestStatus.MULTI_STATUS
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    override fun toXContent(builder: XContentBuilder?, params: ToXContent.Params?): XContentBuilder {
+        return builder!!.startObject()
+            .field(REF_TAG_FIELD, refTag)
+            .field(RECIPIENTS_TAG_FIELD, channelMessageResponseList)
+            .endObject()
+    }
+}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/PluginRestHandler.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/PluginRestHandler.kt
deleted file mode 100644
index e8a17d48..00000000
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/PluginRestHandler.kt
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License").
- * You may not use this file except in compliance with the License.
- * A copy of the License is located at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * or in the "license" file accompanying this file. This file is distributed
- * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
- * express or implied. See the License for the specific language governing
- * permissions and limitations under the License.
- *
- */
-package com.amazon.opendistroforelasticsearch.notifications.resthandler
-
-import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.LOG_PREFIX
-import com.amazon.opendistroforelasticsearch.notifications.util.logger
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.launch
-import org.elasticsearch.client.node.NodeClient
-import org.elasticsearch.rest.BaseRestHandler
-import org.elasticsearch.rest.BaseRestHandler.RestChannelConsumer
-import org.elasticsearch.rest.BytesRestResponse
-import org.elasticsearch.rest.RestChannel
-import org.elasticsearch.rest.RestRequest
-import org.elasticsearch.rest.RestStatus
-import java.io.IOException
-
-/**
- * Plugin Rest handler which executes the request in async scope.
- */
-internal abstract class PluginRestHandler : BaseRestHandler() {
-    internal companion object {
-        protected val scope: CoroutineScope = CoroutineScope(Dispatchers.IO)
-        private val log by logger(PluginRestHandler::class.java)
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer {
-        log.info("$LOG_PREFIX:prepareRequest-${request.method()}=${request.uri()}")
-        return RestChannelConsumer { executeRequestInScope(request, client, it) }
-    }
-
-    /**
-     * Execute the Rest request in scope. To be executed asynchronously
-     * @param request the request to execute
-     * @param client client for executing actions on the local node
-     * @param channel Rest channel to send response to
-     */
-    @Suppress("TooGenericExceptionCaught")
-    private fun executeRequestInScope(request: RestRequest, client: NodeClient, channel: RestChannel) {
-        scope.launch {
-            try {
-                executeRequest(request, client, channel)
-            } catch (exception: IllegalArgumentException) {
-                log.warn("executeRequestInScope:", exception)
-                channel.sendResponse(BytesRestResponse(RestStatus.BAD_REQUEST, exception.message))
-            } catch (exception: IllegalStateException) {
-                log.warn("executeRequestInScope:", exception)
-                channel.sendResponse(BytesRestResponse(RestStatus.SERVICE_UNAVAILABLE, exception.message))
-            } catch (exception: IOException) {
-                log.error("executeRequestInScope: Uncaught IOException:", exception)
-                channel.sendResponse(BytesRestResponse(RestStatus.FAILED_DEPENDENCY, exception.message))
-            } catch (exception: Exception) {
-                log.error("executeRequestInScope: Uncaught Exception:", exception)
-                channel.sendResponse(BytesRestResponse(RestStatus.INTERNAL_SERVER_ERROR, exception.message))
-            }
-        }
-    }
-
-    /**
-     * Execute the Rest request
-     * @param request the request to execute
-     * @param client client for executing actions on the local node
-     * @param channel Rest channel to send response to
-     */
-    abstract fun executeRequest(request: RestRequest, client: NodeClient, channel: RestChannel)
-}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/RestResponseToXContentListener.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/RestResponseToXContentListener.kt
new file mode 100644
index 00000000..983c51a1
--- /dev/null
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/RestResponseToXContentListener.kt
@@ -0,0 +1,36 @@
+/*
+ * Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ *
+ */
+
+package com.amazon.opendistroforelasticsearch.notifications.resthandler
+
+import com.amazon.opendistroforelasticsearch.notifications.model.BaseResponse
+import org.elasticsearch.rest.RestChannel
+import org.elasticsearch.rest.RestStatus
+import org.elasticsearch.rest.action.RestToXContentListener
+
+/**
+ * Overrides RestToXContentListener REST based action listener that assumes the response is of type
+ * {@link ToXContent} and automatically builds an XContent based response
+ * (wrapping the toXContent in startObject/endObject).
+ */
+internal class RestResponseToXContentListener<Response : BaseResponse>(channel: RestChannel) : RestToXContentListener<Response>(channel) {
+    /**
+     * {@inheritDoc}
+     */
+    override fun getStatus(response: Response): RestStatus {
+        return response.getStatus()
+    }
+}
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandler.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandler.kt
similarity index 62%
rename from src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandler.kt
rename to src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandler.kt
index 41f51bc1..427cf775 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandler.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandler.kt
@@ -17,11 +17,14 @@
 package com.amazon.opendistroforelasticsearch.notifications.resthandler
 
 import com.amazon.opendistroforelasticsearch.notifications.NotificationPlugin.Companion.PLUGIN_BASE_URI
-import com.amazon.opendistroforelasticsearch.notifications.action.SendAction
+import com.amazon.opendistroforelasticsearch.notifications.action.SendMessageAction
+import com.amazon.opendistroforelasticsearch.notifications.model.SendMessageRequest
+import com.amazon.opendistroforelasticsearch.notifications.util.contentParserNextToken
 import com.amazon.opendistroforelasticsearch.notifications.util.logger
 import org.elasticsearch.client.node.NodeClient
+import org.elasticsearch.rest.BaseRestHandler
+import org.elasticsearch.rest.BaseRestHandler.RestChannelConsumer
 import org.elasticsearch.rest.BytesRestResponse
-import org.elasticsearch.rest.RestChannel
 import org.elasticsearch.rest.RestHandler.Route
 import org.elasticsearch.rest.RestRequest
 import org.elasticsearch.rest.RestRequest.Method.POST
@@ -31,17 +34,17 @@ import org.elasticsearch.rest.RestStatus
  * Rest handler for sending notification.
  * This handler [SendAction] for sending notification.
  */
-internal class SendRestHandler : PluginRestHandler() {
+internal class SendMessageRestHandler : BaseRestHandler() {
 
     internal companion object {
-        private val log by logger(SendRestHandler::class.java)
+        private val log by logger(SendMessageRestHandler::class.java)
         const val SEND_BASE_URI = "$PLUGIN_BASE_URI/send"
     }
 
     /**
      * {@inheritDoc}
      */
-    override fun getName(): String = "send"
+    override fun getName(): String = "send_message"
 
     /**
      * {@inheritDoc}
@@ -62,11 +65,16 @@ internal class SendRestHandler : PluginRestHandler() {
     /**
      * {@inheritDoc}
      */
-    override fun executeRequest(request: RestRequest, client: NodeClient, channel: RestChannel) {
-        val handler = SendAction(request, client, channel)
-        when (request.method()) {
-            POST -> handler.send()
-            else -> channel.sendResponse(BytesRestResponse(RestStatus.METHOD_NOT_ALLOWED, "${request.method()} is not allowed"))
+    override fun prepareRequest(request: RestRequest, client: NodeClient): RestChannelConsumer {
+        return when (request.method()) {
+            POST -> RestChannelConsumer {
+                client.execute(SendMessageAction.ACTION_TYPE,
+                    SendMessageRequest(request.contentParserNextToken()),
+                    RestResponseToXContentListener(it))
+            }
+            else -> RestChannelConsumer {
+                it.sendResponse(BytesRestResponse(RestStatus.METHOD_NOT_ALLOWED, "${request.method()} is not allowed"))
+            }
         }
     }
 }
diff --git a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/util/Helpers.kt b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/util/Helpers.kt
index 9ec1a58e..7053d45a 100644
--- a/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/util/Helpers.kt
+++ b/src/main/kotlin/com/amazon/opendistroforelasticsearch/notifications/util/Helpers.kt
@@ -18,8 +18,26 @@ package com.amazon.opendistroforelasticsearch.notifications.util
 
 import org.apache.logging.log4j.LogManager
 import org.apache.logging.log4j.Logger
+import org.elasticsearch.common.io.stream.StreamInput
+import org.elasticsearch.common.xcontent.DeprecationHandler
+import org.elasticsearch.common.xcontent.NamedXContentRegistry
+import org.elasticsearch.common.xcontent.ToXContent
+import org.elasticsearch.common.xcontent.ToXContentObject
+import org.elasticsearch.common.xcontent.XContentBuilder
 import org.elasticsearch.common.xcontent.XContentParser
 import org.elasticsearch.common.xcontent.XContentParserUtils
+import org.elasticsearch.common.xcontent.XContentType
+import org.elasticsearch.rest.RestRequest
+
+internal fun StreamInput.createJsonParser(): XContentParser {
+    return XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY, DeprecationHandler.IGNORE_DEPRECATIONS, this)
+}
+
+internal fun RestRequest.contentParserNextToken(): XContentParser {
+    val parser = this.contentParser()
+    parser.nextToken()
+    return parser
+}
 
 internal fun XContentParser.stringList(): List<String> {
     val retList: MutableList<String> = mutableListOf()
@@ -33,3 +51,18 @@ internal fun XContentParser.stringList(): List<String> {
 internal fun <T : Any> logger(forClass: Class<T>): Lazy<Logger> {
     return lazy { LogManager.getLogger(forClass) }
 }
+
+internal fun XContentBuilder.fieldIfNotNull(name: String, value: Any?): XContentBuilder {
+    if (value != null) {
+        this.field(name, value)
+    }
+    return this
+}
+
+internal fun XContentBuilder.objectIfNotNull(name: String, xContentObject: ToXContentObject?): XContentBuilder {
+    if (xContentObject != null) {
+        this.field(name)
+        xContentObject.toXContent(this, ToXContent.EMPTY_PARAMS)
+    }
+    return this
+}
diff --git a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationsRestTestCase.kt b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationsRestTestCase.kt
index 5390f6b8..8e7d9f34 100644
--- a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationsRestTestCase.kt
+++ b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/NotificationsRestTestCase.kt
@@ -16,11 +16,12 @@
 
 package com.amazon.opendistroforelasticsearch.notifications
 
-import com.amazon.opendistroforelasticsearch.notifications.resthandler.SendRestHandler.Companion.SEND_BASE_URI
+import com.amazon.opendistroforelasticsearch.notifications.resthandler.SendMessageRestHandler.Companion.SEND_BASE_URI
 import com.amazon.opendistroforelasticsearch.notifications.settings.PluginSettings
 import com.google.gson.JsonObject
 import org.elasticsearch.client.Request
 import org.elasticsearch.client.RequestOptions
+import org.elasticsearch.client.ResponseException
 import org.junit.After
 import org.junit.Before
 import org.springframework.integration.test.mail.TestMailServer
@@ -71,7 +72,11 @@ abstract class NotificationsRestTestCase : ODFERestTestCase() {
     }
 
     protected fun executeRequest(request: Request): JsonObject {
-        val response = client().performRequest(request)
+        val response = try {
+            client().performRequest(request)
+        } catch (exception: ResponseException) {
+            exception.response
+        }
         val responseBody = getResponseBody(response, true)
         return jsonify(responseBody)
     }
diff --git a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/SmtpChannelIT.kt b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/SmtpChannelIT.kt
index 73c202e5..46af4c9d 100644
--- a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/SmtpChannelIT.kt
+++ b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/channel/SmtpChannelIT.kt
@@ -25,7 +25,7 @@ import com.amazon.opendistroforelasticsearch.notifications.verifyResponse
 import org.elasticsearch.rest.RestStatus
 
 internal class SmtpChannelIT : NotificationsRestTestCase() {
-    private val refTag = "sample raf name"
+    private val refTag = "sample ref name"
     private val title = "sample title"
     private val textDescription = "Description for notification in text"
     private val htmlDescription = "Description for notification in json encode html format"
diff --git a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandlerIT.kt b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandlerIT.kt
similarity index 92%
rename from src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandlerIT.kt
rename to src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandlerIT.kt
index 02d65091..dc1e6133 100644
--- a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandlerIT.kt
+++ b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandlerIT.kt
@@ -18,7 +18,7 @@ package com.amazon.opendistroforelasticsearch.notifications.resthandler
 
 import com.amazon.opendistroforelasticsearch.notifications.ODFERestTestCase
 
-internal class SendRestHandlerIT : ODFERestTestCase() {
+internal class SendMessageRestHandlerIT : ODFERestTestCase() {
 
     fun sendRestHandler() {
     }
diff --git a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandlerTests.kt b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandlerTests.kt
similarity index 86%
rename from src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandlerTests.kt
rename to src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandlerTests.kt
index a70d8caf..5bc3d7d8 100644
--- a/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendRestHandlerTests.kt
+++ b/src/test/kotlin/com/amazon/opendistroforelasticsearch/notifications/resthandler/SendMessageRestHandlerTests.kt
@@ -21,17 +21,17 @@ import org.elasticsearch.rest.RestRequest.Method.POST
 import org.elasticsearch.test.ESTestCase
 import org.junit.jupiter.api.Test
 
-internal class SendRestHandlerTests : ESTestCase() {
+internal class SendMessageRestHandlerTests : ESTestCase() {
 
     @Test
     fun `SendRestHandler name should return send`() {
-        val restHandler = SendRestHandler()
-        assertEquals("send", restHandler.name)
+        val restHandler = SendMessageRestHandler()
+        assertEquals("send_message", restHandler.name)
     }
 
     @Test
     fun `SendRestHandler routes should return send url`() {
-        val restHandler = SendRestHandler()
+        val restHandler = SendMessageRestHandler()
         val routes = restHandler.routes()
         val actualRouteSize = routes.size
         val actualRoute = routes[0]