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 "CProducer.h"
 #include <string.h>
 #include <typeindex>
 #include <string.h>
 #include <typeinfo>
 #include "AsyncCallback.h"
 #include "CBatchMessage.h"
 #include "CCommon.h"
 #include "CMQException.h"
 #include "CMessage.h"
 #include "CSendResult.h"
 #include "DefaultMQProducer.h"
 #include "MQClientErrorContainer.h"

 #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) {
     MQClientErrorContainer::setErr(string(e.what()));
     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) {
     MQClientErrorContainer::setErr(string(e.what()));
     return PRODUCER_SEND_SYNC_FAILED;
   }
   return OK;
 }

 int SendBatchMessage(CProducer* producer, CBatchMessage* batcMsg, CSendResult* result) {
   // CSendResult sendResult;
   if (producer == NULL || batcMsg == NULL || result == NULL) {
     return NULL_POINTER;
   }
   try {
     DefaultMQProducer* defaultMQProducer = (DefaultMQProducer*)producer;
     vector<MQMessage>* message = (vector<MQMessage>*)batcMsg;
     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 (std::type_index(typeid(e)) == std::type_index(typeid(MQException))) {
         MQException& mqe = (MQException&)e;
         cSendCallback->onException(mqe);
       }
       delete cSendCallback;
       cSendCallback = NULL;
     }
     MQClientErrorContainer::setErr(string(e.what()));
     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 SendMessageOnewayOrderly(CProducer* producer, CMessage* msg, QueueSelectorCallback selector, void* arg) {
   if (producer == NULL || msg == NULL) {
     return NULL_POINTER;
   }
   DefaultMQProducer* defaultMQProducer = (DefaultMQProducer*)producer;
   MQMessage* message = (MQMessage*)msg;
   try {
     SelectMessageQueue selectMessageQueue(selector);
     defaultMQProducer->sendOneway(*message, &selectMessageQueue, arg);
   } catch (exception& e) {
     MQClientErrorContainer::setErr(string(e.what()));
     return PRODUCER_SEND_ONEWAY_FAILED;
   }
   return OK;
 }

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

   try {
     // Constructing SelectMessageQueue objects through function pointer callback
     SelectMessageQueue selectMessageQueue(callback);
     defaultMQProducer->send(*message, &selectMessageQueue, arg, cSendCallback);
   } catch (exception& e) {
     printf("%s\n", e.what());
     // std::count<<e.what( )<<std::endl;
     MQClientErrorContainer::setErr(string(e.what()));
     return PRODUCER_SEND_ORDERLYASYNC_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) {
     MQClientErrorContainer::setErr(string(e.what()));
     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 "CProducer.h"
	#include <string.h>
	#include <typeindex>
	#include <string.h>
	#include <typeinfo>
	#include "AsyncCallback.h"
	#include "CBatchMessage.h"
	#include "CCommon.h"
	#include "CMQException.h"
	#include "CMessage.h"
	#include "CSendResult.h"
	#include "DefaultMQProducer.h"
	#include "MQClientErrorContainer.h"

	#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) {
	MQClientErrorContainer::setErr(string(e.what()));
	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) {
	MQClientErrorContainer::setErr(string(e.what()));
	return PRODUCER_SEND_SYNC_FAILED;
	}
	return OK;
	}

	int SendBatchMessage(CProducer* producer, CBatchMessage* batcMsg, CSendResult* result) {
	// CSendResult sendResult;
	if (producer == NULL \|\| batcMsg == NULL \|\| result == NULL) {
	return NULL_POINTER;
	}
	try {
	DefaultMQProducer* defaultMQProducer = (DefaultMQProducer*)producer;
	vector<MQMessage>* message = (vector<MQMessage>*)batcMsg;
	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 (std::type_index(typeid(e)) == std::type_index(typeid(MQException))) {
	MQException& mqe = (MQException&)e;
	cSendCallback->onException(mqe);
	}
	delete cSendCallback;
	cSendCallback = NULL;
	}
	MQClientErrorContainer::setErr(string(e.what()));
	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 SendMessageOnewayOrderly(CProducer* producer, CMessage* msg, QueueSelectorCallback selector, void* arg) {
	if (producer == NULL \|\| msg == NULL) {
	return NULL_POINTER;
	}
	DefaultMQProducer* defaultMQProducer = (DefaultMQProducer*)producer;
	MQMessage* message = (MQMessage*)msg;
	try {
	SelectMessageQueue selectMessageQueue(selector);
	defaultMQProducer->sendOneway(*message, &selectMessageQueue, arg);
	} catch (exception& e) {
	MQClientErrorContainer::setErr(string(e.what()));
	return PRODUCER_SEND_ONEWAY_FAILED;
	}
	return OK;
	}

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

	try {
	// Constructing SelectMessageQueue objects through function pointer callback
	SelectMessageQueue selectMessageQueue(callback);
	defaultMQProducer->send(*message, &selectMessageQueue, arg, cSendCallback);
	} catch (exception& e) {
	printf("%s\n", e.what());
	// std::count<<e.what( )<<std::endl;
	MQClientErrorContainer::setErr(string(e.what()));
	return PRODUCER_SEND_ORDERLYASYNC_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) {
	MQClientErrorContainer::setErr(string(e.what()));
	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