diff --git a/example/MQAdmin.c b/example/MQAdmin.c new file mode 100644 index 000000000..1716a8b95 --- /dev/null +++ b/example/MQAdmin.c @@ -0,0 +1,36 @@ +/* + * 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 +#include "CMQAdmin.h" +#ifdef _WIN32 +#include +#else +#endif + +int main(int argc, char* argv[]) { + CMQAdmin* mqAdmin = CreateMQAdmin("Group_admin"); + printf("MQAdmin initialized. \n"); + SetNamesrvAddrMQAdmin(mqAdmin, "127.0.0.1:9876"); + SetNamesrvDomainMQAdmin(mqAdmin, "rocketmq.nameserver.com"); + StartMQAdmin(mqAdmin); + // add your api here + ShutdownMQAdmin(mqAdmin); + DestroyMQAdmin(mqAdmin); + printf("MQAdmin stopped !\n"); + return 0; +} diff --git a/include/CCommon.h b/include/CCommon.h index ae8a9e1b3..9fda46347 100644 --- a/include/CCommon.h +++ b/include/CCommon.h @@ -50,6 +50,11 @@ typedef enum _CStatus_ { PULLCONSUMER_FETCH_MQ_FAILED = 31, PULLCONSUMER_FETCH_MESSAGE_FAILED = 32, + ADMIN_ERROR_CODE_START = 40, + ADMIN_START_FAILED = 40, + ADMIN_SHUTDOWN_FAILED = 41, + ADMIN_SET_VARIABLE_FAILED = 42, + Not_Support = 500, NOT_SUPPORT_NOW = -1 } CStatus; diff --git a/include/CMQAdmin.h b/include/CMQAdmin.h new file mode 100644 index 000000000..9e82a6dcf --- /dev/null +++ b/include/CMQAdmin.h @@ -0,0 +1,44 @@ +/* + * 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. + */ + +#ifndef __C_MQADMIN_H__ +#define __C_MQADMIN_H__ + +#include "CCommon.h" +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct CMQAdmin CMQAdmin; +ROCKETMQCLIENT_API CMQAdmin* CreateMQAdmin(const char* groupId); +ROCKETMQCLIENT_API int DestroyMQAdmin(CMQAdmin* MQAdmin); +ROCKETMQCLIENT_API int StartMQAdmin(CMQAdmin* MQAdmin); +ROCKETMQCLIENT_API int ShutdownMQAdmin(CMQAdmin* MQAdmin); +ROCKETMQCLIENT_API int SetSessionCredentialsMQAdmin(CMQAdmin* MQAdmin, + const char* accessKey, + const char* secretKey, + const char* channel); +ROCKETMQCLIENT_API int SetNamesrvAddrMQAdmin(CMQAdmin* MQAdmin, const char* addr); +ROCKETMQCLIENT_API int SetNamesrvDomainMQAdmin(CMQAdmin* MQAdmin, const char* domain); +ROCKETMQCLIENT_API int SetMQAdminLogPath(CMQAdmin* MQAdmin, const char* logPath); +ROCKETMQCLIENT_API int SetMQAdminLogFileNumAndSize(CMQAdmin* MQAdmin, int fileNum, long fileSize); +ROCKETMQCLIENT_API int SetMQAdminLogLevel(CMQAdmin* MQAdmin, CLogLevel level); +// you can add some api here +#ifdef __cplusplus +} +#endif +#endif //__C_MQADMIN_H__ diff --git a/src/extern/CMQAdmin.cpp b/src/extern/CMQAdmin.cpp new file mode 100644 index 000000000..22911b5fe --- /dev/null +++ b/src/extern/CMQAdmin.cpp @@ -0,0 +1,158 @@ +/* + * 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 "CMQAdmin.h" +#include "DefaultMQAdmin.h" +#include "MQClientErrorContainer.h" + +#ifdef __cplusplus +extern "C" { +#endif +using namespace rocketmq; +using namespace std; +CMQAdmin* CreateMQAdmin(const char* groupId) { + if (groupId == NULL) { + return NULL; + } + DefaultMQAdmin* defaultMQAdmin = new DefaultMQAdmin(groupId); + return (CMQAdmin*)defaultMQAdmin; +} + +int StartMQAdmin(CMQAdmin* MQAdmin) { + if (MQAdmin == nullptr) { + return NULL_POINTER; + } + DefaultMQAdmin* defaultMQAdmin = (DefaultMQAdmin*)MQAdmin; + try { + defaultMQAdmin->start(); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_START_FAILED; + } + return OK; +} + +int DestroyMQAdmin(CMQAdmin* MQAdmin) { + if (MQAdmin == nullptr) { + return NULL_POINTER; + } + delete reinterpret_cast(MQAdmin); + return OK; +} + +int ShutdownMQAdmin(CMQAdmin* MQAdmin) { + if (MQAdmin == nullptr) { + return NULL_POINTER; + } + DefaultMQAdmin* defaultMQAdmin = (DefaultMQAdmin*)MQAdmin; + try { + defaultMQAdmin->shutdown(); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_SHUTDOWN_FAILED; + } + return OK; +} + +int SetSessionCredentialsMQAdmin(CMQAdmin* MQAdmin, const char* accessKey, const char* secretKey, const char* channel) { + if (MQAdmin == nullptr) { + return NULL_POINTER; + } + DefaultMQAdmin* defaultMQAdmin = (DefaultMQAdmin*)MQAdmin; + try { + defaultMQAdmin->setSessionCredentials(accessKey, secretKey, channel); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_SET_VARIABLE_FAILED; + } + return OK; +} + +int SetNamesrvAddrMQAdmin(CMQAdmin* MQAdmin, const char* addr) { + if (MQAdmin == nullptr) { + return NULL_POINTER; + } + DefaultMQAdmin* defaultMQAdmin = (DefaultMQAdmin*)MQAdmin; + try { + defaultMQAdmin->setNamesrvAddr(addr); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_SET_VARIABLE_FAILED; + } + return OK; +} + +int SetNamesrvDomainMQAdmin(CMQAdmin* MQAdmin, const char* domain) { + if (MQAdmin == nullptr) { + return NULL_POINTER; + } + DefaultMQAdmin* defaultMQAdmin = (DefaultMQAdmin*)MQAdmin; + try { + defaultMQAdmin->setNamesrvDomain(domain); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_SET_VARIABLE_FAILED; + } + return OK; +} + +int SetMQAdminLogPath(CMQAdmin* MQAdmin, const char* logPath) { + if (MQAdmin == NULL) { + return NULL_POINTER; + } + try { + setenv(ROCKETMQ_CLIENT_LOG_DIR.c_str(), logPath, 1); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_SET_VARIABLE_FAILED; + } + return OK; +} + +int SetMQAdminLogFileNumAndSize(CMQAdmin* MQAdmin, int fileNum, long fileSize) { + if (MQAdmin == NULL) { + return NULL_POINTER; + } + if (fileNum <= 0 || fileSize <= 0) { + return ADMIN_SET_VARIABLE_FAILED; + } + DefaultMQAdmin* defaultMQAdmin = (DefaultMQAdmin*)MQAdmin; + try { + defaultMQAdmin->setLogFileSizeAndNum(fileNum, fileSize); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_SET_VARIABLE_FAILED; + } + return OK; +} + +int SetMQAdminLogLevel(CMQAdmin* MQAdmin, CLogLevel level) { + if (MQAdmin == NULL) { + return NULL_POINTER; + } + DefaultMQAdmin* defaultMQAdmin = (DefaultMQAdmin*)MQAdmin; + try { + defaultMQAdmin->setLogLevel((elogLevel)level); + } catch (exception& e) { + MQClientErrorContainer::setErr(string(e.what())); + return ADMIN_SET_VARIABLE_FAILED; + } + return OK; +} + +#ifdef __cplusplus +}; +#endif \ No newline at end of file diff --git a/test/src/client/DefaultMQAdminTest.cpp b/test/src/client/DefaultMQAdminTest.cpp new file mode 100644 index 000000000..401193bc4 --- /dev/null +++ b/test/src/client/DefaultMQAdminTest.cpp @@ -0,0 +1,56 @@ +/* + * 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 +#include + +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +#include "DefaultMQAdmin.h" + +using namespace std; +using namespace rocketmq; +using rocketmq::DefaultMQAdmin; +using testing::_; +using ::testing::InitGoogleMock; +using ::testing::InitGoogleTest; +using testing::Return; + +TEST(DefaultMQAdminTest, init) { + DefaultMQAdmin* impl = new DefaultMQAdmin("testMQAdminGroup"); + EXPECT_EQ(impl->getGroupName(), "testMQAdminGroup"); + impl->setUnitName("testUnit"); + EXPECT_EQ(impl->getUnitName(), "testUnit"); + impl->setTcpTransportPullThreadNum(64); + EXPECT_EQ(impl->getTcpTransportPullThreadNum(), 64); + impl->setTcpTransportConnectTimeout(2000); + EXPECT_EQ(impl->getTcpTransportConnectTimeout(), 2000); + impl->setTcpTransportTryLockTimeout(3000); + EXPECT_EQ(impl->getTcpTransportTryLockTimeout(), 3); + impl->setNamesrvAddr("http://rocketmq.nameserver.com"); + EXPECT_EQ(impl->getNamesrvAddr(), "rocketmq.nameserver.com"); + impl->setNameSpace("MQ_INST_NAMESPACE_TEST"); + EXPECT_EQ(impl->getNameSpace(), "MQ_INST_NAMESPACE_TEST"); + impl->setMessageTrace(true); + EXPECT_TRUE(impl->getMessageTrace()); +} + +int main(int argc, char* argv[]) { + InitGoogleMock(&argc, argv); + return RUN_ALL_TESTS(); +}