| /* |
| * 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 "ProducerImpl.h" |
| #include "rocketmq.h" |
| #include "rocketmq-ons-cpp-full.h" |
| #include "SendResultONS.h" |
| #include "Logger.h" |
| #include "common/UtilAll.h" |
| |
| using namespace ons; |
| |
| ProducerImpl::ProducerImpl() throw(ONSClientException) { |
| } |
| |
| ProducerImpl::ProducerImpl(ONSFactoryProperty factoryProperty) throw(ons::ONSClientException) { |
| std::string logPath(factoryProperty.getLogPath()); |
| rocketmq::spd_log::initLogger(logPath, rocketmq::LOGLEVEL_INFO); |
| graal_isolatethread_t *thread_; |
| ThreadAttachment attachment(&thread_); |
| factory_property fp; |
| FactoryPropertyConverter converter(factoryProperty, fp); |
| instanceIndex_ = create_producer(thread_, &fp); |
| rocketmq::spd_log::info("Create Producer OK, InstanceId:{}, ProducerID:{}, NameServer:{}", |
| instanceIndex_, factoryProperty.getProducerId(), factoryProperty.getNameSrvAddr()); |
| } |
| |
| ProducerImpl::~ProducerImpl() { |
| rocketmq::spd_log::destoryLogger(); |
| } |
| |
| void ProducerImpl::start() { |
| |
| } |
| |
| void ProducerImpl::shutdown() { |
| graal_isolatethread_t *thread_; |
| ThreadAttachment attachment(&thread_); |
| destroy_instance(thread_, instanceIndex_); |
| rocketmq::spd_log::info("Destroy Producer instance {} OK", instanceIndex_); |
| } |
| |
| ons::SendResultONS ProducerImpl::send(Message &msg) throw(ONSClientException) { |
| message m; |
| MessageConverter converter(m, msg); |
| send_result sendResult; |
| SendResultWrapper wrapper(sendResult); |
| graal_isolatethread_t *thread_; |
| ThreadAttachment attachment(&thread_); |
| send_message(thread_, instanceIndex_, &m, &sendResult); |
| if (sendResult.error_no) { |
| throw ONSClientException(std::string(sendResult.error_msg), sendResult.error_no); |
| } |
| rocketmq::spd_log::debug("Send message OK. MsgId: {}", sendResult.message_id); |
| SendResultONS sendResultOns; |
| sendResultOns.setMessageId(std::string(sendResult.message_id)); |
| return sendResultOns; |
| } |
| |
| SendResultONS ProducerImpl::send(Message &msg, const MessageQueueONS &mq) throw(ons::ONSClientException) { |
| return send(msg); |
| } |
| |
| #ifdef __cplusplus |
| extern "C" { |
| #endif |
| void on_success_func(void *thread, char *message_id, char *send_callback_ons) { |
| auto sendCallbackONS = reinterpret_cast<SendCallbackONS *>(send_callback_ons); |
| SendResultONS sendResultOns; |
| sendResultOns.setMessageId(message_id); |
| sendCallbackONS->onSuccess(sendResultOns); |
| } |
| |
| void on_exception_func(void *thread, char *m_msg, int m_error, char *send_callback_ons) { |
| auto sendCallbackONS = reinterpret_cast<SendCallbackONS *>(send_callback_ons); |
| ONSClientException onsClientException(m_msg, m_error); |
| sendCallbackONS->onException(onsClientException); |
| } |
| #ifdef __cplusplus |
| } |
| #endif |
| |
| void ProducerImpl::sendAsync(Message &msg, SendCallbackONS *pSendCallback) throw(ons::ONSClientException) { |
| message m; |
| MessageConverter converter(m, msg); |
| send_result sendResult; |
| SendResultWrapper wrapper(sendResult); |
| callback_func c_f; |
| if (pSendCallback == NULL) { |
| std::string msg = "Send Callback cannot be NULL."; |
| throw ONSClientException(msg, -1); |
| } |
| c_f.on_success = on_success_func; |
| c_f.on_exception = on_exception_func; |
| c_f.send_callback_ons = reinterpret_cast<char *>(pSendCallback); |
| |
| graal_isolatethread_t *thread_; |
| ThreadAttachment attachment(&thread_); |
| send_message_async(thread_, instanceIndex_, &m, &sendResult, &c_f); |
| if (sendResult.error_no) { |
| throw ONSClientException(std::string(sendResult.error_msg), sendResult.error_no); |
| } |
| } |
| |
| void ProducerImpl::sendOneway(Message &msg) throw(ons::ONSClientException) { |
| message m; |
| MessageConverter converter(m, msg); |
| send_result sendResult; |
| bzero(&sendResult, sizeof(send_result)); |
| SendResultWrapper wrapper(sendResult); |
| graal_isolatethread_t *thread_; |
| ThreadAttachment attachment(&thread_); |
| send_message_oneway(thread_, instanceIndex_, &m, &sendResult); |
| if (sendResult.error_no) { |
| throw ONSClientException(std::string(sendResult.error_msg), sendResult.error_no); |
| } |
| } |