src/extern/CProducer.cpp - rocketmq-client-cpp - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License 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.
  */

 #include "DefaultMQProducer.h"
 #include "AsyncCallback.h"
 #include "CProducer.h"
 #include "CCommon.h"
 #include "CSendResult.h"
 #include "CMessage.h"
 #include "CMQException.h"
 #include <string.h>
 #include <typeinfo>

 #ifdef __cplusplus
 extern "C" {
 #endif
 using namespace rocketmq;
 using namespace std;

 class SelectMessageQueue : public MessageQueueSelector {
 public:
     SelectMessageQueue(QueueSelectorCallback callback) {
         m_pCallback = callback;
     }

     MQMessageQueue select(const std::vector<MQMessageQueue> &mqs,
                           const MQMessage &msg, void *arg) {
         CMessage *message = (CMessage *) &msg;
         //Get the index of sending MQMessageQueue through callback function.
         int index = m_pCallback(mqs.size(), message, arg);
         return mqs[index];
     }

 private:
     QueueSelectorCallback m_pCallback;
 };

 class CSendCallback : public AutoDeleteSendCallBack{
 public:
     CSendCallback(CSendSuccessCallback cSendSuccessCallback,CSendExceptionCallback cSendExceptionCallback){
         m_cSendSuccessCallback = cSendSuccessCallback;
         m_cSendExceptionCallback = cSendExceptionCallback;
     }
     virtual ~CSendCallback(){}
     virtual void onSuccess(SendResult& sendResult) {
         CSendResult result;
         result.sendStatus = CSendStatus((int) sendResult.getSendStatus());
         result.offset = sendResult.getQueueOffset();
         strncpy(result.msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
         result.msgId[MAX_MESSAGE_ID_LENGTH - 1] = 0;
         m_cSendSuccessCallback(result);
     }
     virtual void onException(MQException& e) {
         CMQException exception;
         exception.error = e.GetError();
         exception.line  = e.GetLine();
         strncpy(exception.msg, e.what(), MAX_EXEPTION_MSG_LENGTH - 1);
         strncpy(exception.file, e.GetFile(), MAX_EXEPTION_FILE_LENGTH - 1);
         m_cSendExceptionCallback( exception );
     }
 private:
     CSendSuccessCallback m_cSendSuccessCallback;
     CSendExceptionCallback m_cSendExceptionCallback;
 };

 CProducer *CreateProducer(const char *groupId) {
     if (groupId == NULL) {
         return NULL;
     }
     DefaultMQProducer *defaultMQProducer = new DefaultMQProducer(groupId);
     return (CProducer *) defaultMQProducer;
 }
 int DestroyProducer(CProducer *pProducer) {
     if (pProducer == NULL) {
         return NULL_POINTER;
     }
     delete reinterpret_cast<DefaultMQProducer * >(pProducer);
     return OK;
 }
 int StartProducer(CProducer *producer) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     try {
         ((DefaultMQProducer *) producer)->start();
     } catch (exception &e) {
         return PRODUCER_START_FAILED;
     }
     return OK;
 }
 int ShutdownProducer(CProducer *producer) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->shutdown();
     return OK;
 }
 int SetProducerNameServerAddress(CProducer *producer, const char *namesrv) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setNamesrvAddr(namesrv);
     return OK;
 }
 int SetProducerNameServerDomain(CProducer *producer, const char *domain) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setNamesrvDomain(domain);
     return OK;
 }
 int SendMessageSync(CProducer *producer, CMessage *msg, CSendResult *result) {
     //CSendResult sendResult;
     if (producer == NULL || msg == NULL || result == NULL) {
         return NULL_POINTER;
     }
     try {
         DefaultMQProducer *defaultMQProducer = (DefaultMQProducer *) producer;
         MQMessage *message = (MQMessage *) msg;
         SendResult sendResult = defaultMQProducer->send(*message);
         switch (sendResult.getSendStatus()) {
             case SEND_OK:
                 result->sendStatus = E_SEND_OK;
                 break;
             case SEND_FLUSH_DISK_TIMEOUT:
                 result->sendStatus = E_SEND_FLUSH_DISK_TIMEOUT;
                 break;
             case SEND_FLUSH_SLAVE_TIMEOUT:
                 result->sendStatus = E_SEND_FLUSH_SLAVE_TIMEOUT;
                 break;
             case SEND_SLAVE_NOT_AVAILABLE:
                 result->sendStatus = E_SEND_SLAVE_NOT_AVAILABLE;
                 break;
             default:
                 result->sendStatus = E_SEND_OK;
                 break;
         }
         result->offset = sendResult.getQueueOffset();
         strncpy(result->msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
         result->msgId[MAX_MESSAGE_ID_LENGTH - 1] = 0;
     } catch (exception &e) {
         return PRODUCER_SEND_SYNC_FAILED;
     }
     return OK;
 }

 int SendMessageAsync(CProducer *producer, CMessage *msg, CSendSuccessCallback cSendSuccessCallback, CSendExceptionCallback cSendExceptionCallback){
     if (producer == NULL || msg == NULL || cSendSuccessCallback == NULL || cSendExceptionCallback == NULL) {
         return NULL_POINTER;
     }
     DefaultMQProducer *defaultMQProducer = (DefaultMQProducer *) producer;
     MQMessage *message = (MQMessage *) msg;
     CSendCallback* cSendCallback = new CSendCallback(cSendSuccessCallback , cSendExceptionCallback);

     try {
         defaultMQProducer->send(*message ,cSendCallback);
     } catch (exception &e) {
         if(cSendCallback != NULL){
             if(typeid(e) == typeid( MQException )){
                 MQException &mqe = (MQException &)e;
                 cSendCallback->onException( mqe );
             }
             delete cSendCallback;
             cSendCallback = NULL;
         }
         return PRODUCER_SEND_ASYNC_FAILED;
     }
     return OK;
 }

 int SendMessageOneway(CProducer *producer, CMessage *msg) {
     if (producer == NULL || msg == NULL) {
         return NULL_POINTER;
     }
     DefaultMQProducer *defaultMQProducer = (DefaultMQProducer *) producer;
     MQMessage *message = (MQMessage *) msg;
     try {
         defaultMQProducer->sendOneway(*message);
     } catch (exception &e) {
         return PRODUCER_SEND_ONEWAY_FAILED;
     }
     return OK;
 }

 int
 SendMessageOrderly(CProducer *producer, CMessage *msg, QueueSelectorCallback callback, void *arg, int autoRetryTimes,
                    CSendResult *result) {
     if (producer == NULL || msg == NULL || callback == NULL || arg == NULL || result == NULL) {
         return NULL_POINTER;
     }
     DefaultMQProducer *defaultMQProducer = (DefaultMQProducer *) producer;
     MQMessage *message = (MQMessage *) msg;
     try {
         //Constructing SelectMessageQueue objects through function pointer callback
         SelectMessageQueue selectMessageQueue(callback);
         SendResult sendResult = defaultMQProducer->send(*message, &selectMessageQueue, arg, autoRetryTimes);
         //Convert SendStatus to CSendStatus
         result->sendStatus = CSendStatus((int) sendResult.getSendStatus());
         result->offset = sendResult.getQueueOffset();
         strncpy(result->msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
         result->msgId[MAX_MESSAGE_ID_LENGTH - 1] = 0;
     } catch (exception &e) {
         return PRODUCER_SEND_ORDERLY_FAILED;
     }
     return OK;
 }

 int SetProducerGroupName(CProducer *producer, const char *groupName) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setGroupName(groupName);
     return OK;
 }
 int SetProducerInstanceName(CProducer *producer, const char *instanceName) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setInstanceName(instanceName);
     return OK;
 }
 int SetProducerSessionCredentials(CProducer *producer, const char *accessKey, const char *secretKey,
                                   const char *onsChannel) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setSessionCredentials(accessKey, secretKey, onsChannel);
     return OK;
 }
 int SetProducerLogPath(CProducer *producer, const char *logPath) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     //Todo, This api should be implemented by core api.
     //((DefaultMQProducer *) producer)->setLogFileSizeAndNum(3, 102400000);
     return OK;
 }

 int SetProducerLogFileNumAndSize(CProducer *producer, int fileNum, long fileSize) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setLogFileSizeAndNum(fileNum, fileSize);
     return OK;
 }

 int SetProducerLogLevel(CProducer *producer, CLogLevel level) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setLogLevel((elogLevel) level);
     return OK;
 }

 int SetProducerSendMsgTimeout(CProducer *producer, int timeout) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setSendMsgTimeout(timeout);
     return OK;
 }

 int SetProducerCompressMsgBodyOverHowmuch(CProducer *producer, int howmuch) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setCompressMsgBodyOverHowmuch(howmuch);
     return OK;
 }

 int SetProducerCompressLevel(CProducer *producer, int level) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setCompressLevel(level);
     return OK;
 }

 int SetProducerMaxMessageSize(CProducer *producer, int size) {
     if (producer == NULL) {
         return NULL_POINTER;
     }
     ((DefaultMQProducer *) producer)->setMaxMessageSize(size);
     return OK;
 }
 #ifdef __cplusplus
 };
 #endif
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License 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.
	*/

	#include "DefaultMQProducer.h"
	#include "AsyncCallback.h"
	#include "CProducer.h"
	#include "CCommon.h"
	#include "CSendResult.h"
	#include "CMessage.h"
	#include "CMQException.h"
	#include <string.h>
	#include <typeinfo>

	#ifdef __cplusplus
	extern "C" {
	#endif
	using namespace rocketmq;
	using namespace std;

	class SelectMessageQueue : public MessageQueueSelector {
	public:
	SelectMessageQueue(QueueSelectorCallback callback) {
	m_pCallback = callback;
	}

	MQMessageQueue select(const std::vector<MQMessageQueue> &mqs,
	const MQMessage &msg, void *arg) {
	CMessage message = (CMessage ) &msg;
	//Get the index of sending MQMessageQueue through callback function.
	int index = m_pCallback(mqs.size(), message, arg);
	return mqs[index];
	}

	private:
	QueueSelectorCallback m_pCallback;
	};

	class CSendCallback : public AutoDeleteSendCallBack{
	public:
	CSendCallback(CSendSuccessCallback cSendSuccessCallback,CSendExceptionCallback cSendExceptionCallback){
	m_cSendSuccessCallback = cSendSuccessCallback;
	m_cSendExceptionCallback = cSendExceptionCallback;
	}
	virtual ~CSendCallback(){}
	virtual void onSuccess(SendResult& sendResult) {
	CSendResult result;
	result.sendStatus = CSendStatus((int) sendResult.getSendStatus());
	result.offset = sendResult.getQueueOffset();
	strncpy(result.msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
	result.msgId[MAX_MESSAGE_ID_LENGTH - 1] = 0;
	m_cSendSuccessCallback(result);
	}
	virtual void onException(MQException& e) {
	CMQException exception;
	exception.error = e.GetError();
	exception.line = e.GetLine();
	strncpy(exception.msg, e.what(), MAX_EXEPTION_MSG_LENGTH - 1);
	strncpy(exception.file, e.GetFile(), MAX_EXEPTION_FILE_LENGTH - 1);
	m_cSendExceptionCallback( exception );
	}
	private:
	CSendSuccessCallback m_cSendSuccessCallback;
	CSendExceptionCallback m_cSendExceptionCallback;
	};

	CProducer CreateProducer(const char groupId) {
	if (groupId == NULL) {
	return NULL;
	}
	DefaultMQProducer *defaultMQProducer = new DefaultMQProducer(groupId);
	return (CProducer *) defaultMQProducer;
	}
	int DestroyProducer(CProducer *pProducer) {
	if (pProducer == NULL) {
	return NULL_POINTER;
	}
	delete reinterpret_cast<DefaultMQProducer * >(pProducer);
	return OK;
	}
	int StartProducer(CProducer *producer) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	try {
	((DefaultMQProducer *) producer)->start();
	} catch (exception &e) {
	return PRODUCER_START_FAILED;
	}
	return OK;
	}
	int ShutdownProducer(CProducer *producer) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->shutdown();
	return OK;
	}
	int SetProducerNameServerAddress(CProducer producer, const char namesrv) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setNamesrvAddr(namesrv);
	return OK;
	}
	int SetProducerNameServerDomain(CProducer producer, const char domain) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setNamesrvDomain(domain);
	return OK;
	}
	int SendMessageSync(CProducer producer, CMessage msg, CSendResult *result) {
	//CSendResult sendResult;
	if (producer == NULL \|\| msg == NULL \|\| result == NULL) {
	return NULL_POINTER;
	}
	try {
	DefaultMQProducer defaultMQProducer = (DefaultMQProducer ) producer;
	MQMessage message = (MQMessage ) msg;
	SendResult sendResult = defaultMQProducer->send(*message);
	switch (sendResult.getSendStatus()) {
	case SEND_OK:
	result->sendStatus = E_SEND_OK;
	break;
	case SEND_FLUSH_DISK_TIMEOUT:
	result->sendStatus = E_SEND_FLUSH_DISK_TIMEOUT;
	break;
	case SEND_FLUSH_SLAVE_TIMEOUT:
	result->sendStatus = E_SEND_FLUSH_SLAVE_TIMEOUT;
	break;
	case SEND_SLAVE_NOT_AVAILABLE:
	result->sendStatus = E_SEND_SLAVE_NOT_AVAILABLE;
	break;
	default:
	result->sendStatus = E_SEND_OK;
	break;
	}
	result->offset = sendResult.getQueueOffset();
	strncpy(result->msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
	result->msgId[MAX_MESSAGE_ID_LENGTH - 1] = 0;
	} catch (exception &e) {
	return PRODUCER_SEND_SYNC_FAILED;
	}
	return OK;
	}

	int SendMessageAsync(CProducer producer, CMessage msg, CSendSuccessCallback cSendSuccessCallback, CSendExceptionCallback cSendExceptionCallback){
	if (producer == NULL \|\| msg == NULL \|\| cSendSuccessCallback == NULL \|\| cSendExceptionCallback == NULL) {
	return NULL_POINTER;
	}
	DefaultMQProducer defaultMQProducer = (DefaultMQProducer ) producer;
	MQMessage message = (MQMessage ) msg;
	CSendCallback* cSendCallback = new CSendCallback(cSendSuccessCallback , cSendExceptionCallback);

	try {
	defaultMQProducer->send(*message ,cSendCallback);
	} catch (exception &e) {
	if(cSendCallback != NULL){
	if(typeid(e) == typeid( MQException )){
	MQException &mqe = (MQException &)e;
	cSendCallback->onException( mqe );
	}
	delete cSendCallback;
	cSendCallback = NULL;
	}
	return PRODUCER_SEND_ASYNC_FAILED;
	}
	return OK;
	}

	int SendMessageOneway(CProducer producer, CMessage msg) {
	if (producer == NULL \|\| msg == NULL) {
	return NULL_POINTER;
	}
	DefaultMQProducer defaultMQProducer = (DefaultMQProducer ) producer;
	MQMessage message = (MQMessage ) msg;
	try {
	defaultMQProducer->sendOneway(*message);
	} catch (exception &e) {
	return PRODUCER_SEND_ONEWAY_FAILED;
	}
	return OK;
	}

	int
	SendMessageOrderly(CProducer producer, CMessage msg, QueueSelectorCallback callback, void *arg, int autoRetryTimes,
	CSendResult *result) {
	if (producer == NULL \|\| msg == NULL \|\| callback == NULL \|\| arg == NULL \|\| result == NULL) {
	return NULL_POINTER;
	}
	DefaultMQProducer defaultMQProducer = (DefaultMQProducer ) producer;
	MQMessage message = (MQMessage ) msg;
	try {
	//Constructing SelectMessageQueue objects through function pointer callback
	SelectMessageQueue selectMessageQueue(callback);
	SendResult sendResult = defaultMQProducer->send(*message, &selectMessageQueue, arg, autoRetryTimes);
	//Convert SendStatus to CSendStatus
	result->sendStatus = CSendStatus((int) sendResult.getSendStatus());
	result->offset = sendResult.getQueueOffset();
	strncpy(result->msgId, sendResult.getMsgId().c_str(), MAX_MESSAGE_ID_LENGTH - 1);
	result->msgId[MAX_MESSAGE_ID_LENGTH - 1] = 0;
	} catch (exception &e) {
	return PRODUCER_SEND_ORDERLY_FAILED;
	}
	return OK;
	}

	int SetProducerGroupName(CProducer producer, const char groupName) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setGroupName(groupName);
	return OK;
	}
	int SetProducerInstanceName(CProducer producer, const char instanceName) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setInstanceName(instanceName);
	return OK;
	}
	int SetProducerSessionCredentials(CProducer producer, const char accessKey, const char *secretKey,
	const char *onsChannel) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setSessionCredentials(accessKey, secretKey, onsChannel);
	return OK;
	}
	int SetProducerLogPath(CProducer producer, const char logPath) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	//Todo, This api should be implemented by core api.
	//((DefaultMQProducer *) producer)->setLogFileSizeAndNum(3, 102400000);
	return OK;
	}

	int SetProducerLogFileNumAndSize(CProducer *producer, int fileNum, long fileSize) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setLogFileSizeAndNum(fileNum, fileSize);
	return OK;
	}

	int SetProducerLogLevel(CProducer *producer, CLogLevel level) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setLogLevel((elogLevel) level);
	return OK;
	}

	int SetProducerSendMsgTimeout(CProducer *producer, int timeout) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setSendMsgTimeout(timeout);
	return OK;
	}

	int SetProducerCompressMsgBodyOverHowmuch(CProducer *producer, int howmuch) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setCompressMsgBodyOverHowmuch(howmuch);
	return OK;
	}

	int SetProducerCompressLevel(CProducer *producer, int level) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setCompressLevel(level);
	return OK;
	}

	int SetProducerMaxMessageSize(CProducer *producer, int size) {
	if (producer == NULL) {
	return NULL_POINTER;
	}
	((DefaultMQProducer *) producer)->setMaxMessageSize(size);
	return OK;
	}
	#ifdef __cplusplus
	};
	#endif