[ISSUE #494] [Part-B] Expose the MqAdmin API

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 <stdio.h>
+#include "CMQAdmin.h"
+#ifdef _WIN32
+#include <windows.h>
+#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;
+}

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;

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__

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<DefaultMQAdmin*>(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

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 <map>
+#include <vector>
+
+#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();
+}