messageStructure = request.getParameter("MessageStructure");
if (!messageStructure.equals("json")) {
logger.error("event=cns_publish error_code=InvalidParameters message=" + message + " message_structure=" + messageStructure + " topic_arn=" + topicArn + " user_id=" + userId);
throw new CMBException(CNSErrorCodes.CNS_InvalidParameter,"Invalid parameter: Invalid Message Structure parameter: " + messageStructure);
}
cnsMessage.setMessageStructure(CNSMessage.CNSMessageStructure.valueOf(messageStructure));
}
if (request.getParameter("Subject") != null) {
subject = request.getParameter("Subject");
cnsMessage.setSubject(subject);
}
if ((userId == null) || (message == null)) {
logger.error("event=cns_publish error_code=InvalidParameters message=" + message + " topic_arn=" + topicArn + " user_id=" + userId);
throw new CMBException(CNSErrorCodes.CNS_ValidationError,"1 validation error detected: Value null at 'message' failed to satisfy constraint: Member must not be null");
}
if ((topicArn == null) || !Util.isValidTopicArn(topicArn)) {
logger.error("event=cns_publish error_code=InvalidParameters message=" + message + " topic_arn=" + topicArn + " user_id=" + userId);
throw new CMBException(CNSErrorCodes.CNS_InvalidParameter,"TopicArn");
}
CNSTopic topic = CNSCache.getTopic(topicArn);
if (topic == null) {
logger.error("event=cns_publish error_code=NotFound message=" + message + " topic_arn=" + topicArn + " user_id=" + userId);
throw new CMBException(CNSErrorCodes.CNS_NotFound,"Resource not found.");
}
cnsMessage.setUserId(topic.getUserId());
cnsMessage.setTopicArn(topicArn);
cnsMessage.setMessageType(CNSMessageType.Notification);
cnsMessage.checkIsValid();
CNSTopicAttributes topicAttributes = CNSCache.getTopicAttributes(topicArn);
List<String> receiptHandles = new ArrayList<String>();
boolean success = true;
if (topicAttributes != null && topicAttributes.getSubscriptionsConfirmed() == 0) {
// optimization: don't do anything if there are no confirmed subscribers
logger.warn("event=no_confirmed_subscribers action=publish topic_arn=" + topicArn);
} else if (CMBProperties.getInstance().isCNSBypassPublishJobQueueForSmallTopics() && topicAttributes != null && topicAttributes.getSubscriptionsConfirmed() <= CMBProperties.getInstance().getCNSMaxSubscriptionsPerEndpointPublishJob()) {
// optimization: if there's only one chunk due to few subscribers, write directly into endpoint publish queue bypassing the publish job queue
logger.debug("event=using_job_queue_overpass");
List<CNSEndpointPublishJob.CNSEndpointSubscriptionInfo> subscriptions = CNSEndpointPublisherJobProducer.getSubscriptionsForTopic(topicArn);
if (subscriptions != null && subscriptions.size() > 0) {
List<CNSEndpointPublishJob> epPublishJobs = CNSEndpointPublisherJobProducer.createEndpointPublishJobs(cnsMessage, subscriptions);
if (epPublishJobs.size() != 1) {
logger.warn("event=unexpected_number_of_endpoint_publish_jobs count=" + epPublishJobs.size());
}
for (CNSEndpointPublishJob epPublishJob: epPublishJobs) {
String handle = sendMessageOnRandomShardAndCreateQueueIfAbsent(CNS_ENDPOINT_QUEUE_NAME_PREFIX, CMBProperties.getInstance().getCNSNumEndpointPublishJobQueues(), epPublishJob.serialize(), cnsInternalUser.getUserId());
if (handle != null && !handle.equals("")) {
receiptHandles.add(handle);
} else {
success = false;
}
}
}
} else {
// otherwise pick publish job queue
logger.debug("event=going_through_job_queue_town_center");
String handle = sendMessageOnRandomShardAndCreateQueueIfAbsent(CNS_PUBLISH_QUEUE_NAME_PREFIX, CMBProperties.getInstance().getCNSNumPublishJobQueues(), cnsMessage.serialize(), cnsInternalUser.getUserId());
if (handle != null && !handle.equals("")) {
receiptHandles.add(handle);
} else {
success = false;
}
}
if (!success) {
throw new CMBException(CMBErrorCodes.InternalError, "Failed to place message on internal cns queue");
}
request.setReceiptHandles(receiptHandles);
String out = CNSSubscriptionPopulator.getPublishResponse(receiptHandles);