[agent] start rabbitmq , continue a lot

server/src/main/java/io/wdd/rpc/init/AcceptBootUpInfoMessage.java

@@ -0,0 +1,82 @@
+package io.wdd.rpc.init;
+
+
+import com.fasterxml.jackson.databind.json.JsonMapper;
+import io.wdd.common.beans.rabbitmq.OctopusMessage;
+import io.wdd.common.handler.MyRuntimeException;
+import io.wdd.common.utils.MessageUtils;
+import io.wdd.server.beans.po.ServerInfoPO;
+import io.wdd.server.beans.vo.ServerInfoVO;
+import io.wdd.server.utils.DaemonDatabaseOperator;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.*;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.io.IOException;
+
+/**
+ * The type Accept boot up info message.
+ */
+@Service
+public class AcceptBootUpInfoMessage {
+
+
+    @Resource
+    DaemonDatabaseOperator databaseOperator;
+
+    /**
+     * Handle octopus agent boot up info.
+     *
+     * @param message the message
+     */
+    @RabbitHandler
+    @RabbitListener(
+            bindings =
+            @QueueBinding(
+                    value = @Queue(name = "${octopus.message.init_to_server}"),
+                    exchange = @Exchange(name = "${octopus.message.init_exchange}", type = "direct"),
+                    key = {"${octopus.message.init_to_server_key}"}
+            )
+            ,
+            ackMode = "MANUAL"
+    )
+    public void handleOctopusAgentBootUpInfo(Message message) {
+
+        JsonMapper jsonMapper = new JsonMapper();
+        ServerInfoVO serverInfoVO;
+
+        try {
+            serverInfoVO = jsonMapper.readValue(message.getBody(), ServerInfoVO.class);
+        } catch (IOException e) {
+            throw new MyRuntimeException("parse rabbit server info error, please check !");
+        }
+
+
+        // 1. check if information is correct
+        if(!validateServerInfo(serverInfoVO)){
+            throw new MyRuntimeException("server info validated failed !");
+        };
+        // 2. generate the unique topic for agent
+        String agentQueueTopic = generateAgentQueueTopic(serverInfoVO);
+        // 3. save the agent info into database
+        // backend fixed thread daemon to operate the database ensuring the operation is correct !
+        if(!databaseOperator.saveInitOctopusAgentInfo(serverInfoVO)){
+            throw new MyRuntimeException("database save agent info error !");
+        }
+
+        // 4. send InitMessage to agent
+
+    }
+
+    private String generateAgentQueueTopic(ServerInfoVO serverInfoVO) {
+        return null;
+    }
+
+    private boolean validateServerInfo(ServerInfoVO serverInfoVO) {
+
+        return false;
+    }
+
+
+}

server/src/main/java/io/wdd/rpc/init/FromServerMessageBinding.java

@@ -0,0 +1,53 @@
+package io.wdd.rpc.init;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class FromServerMessageBinding {
+
+    @Value("${octopus.message.init_exchange}")
+    public String INIT_EXCHANGE;
+
+    @Value("${octopus.message.init_from_server}")
+    public String INIT_FROM_SERVER;
+
+    @Value("${octopus.message.init_to_server}")
+    public String INIT_TO_SERVER;
+
+    @Value("${octopus.message.init_from_server_key}")
+    public String INIT_FROM_SERVER_KEY;
+
+    @Value("${octopus.message.init_to_server_key}")
+    public String INIT_TO_SERVER_KEY;
+
+    @Bean
+    public DirectExchange initDirectExchange() {
+        return new DirectExchange(INIT_EXCHANGE);
+    }
+
+    @Bean
+    public Queue initFromServerQueue() {
+        return new Queue(INIT_FROM_SERVER);
+    }
+
+
+    /**
+     * 配置一个队列和交换机的绑定
+     *
+     * @param initDirectQueue    : 需要绑定的队列对象，参数名必须和某个@Bean的方法名完全相同，这样就会进行自动注入，对应 .bind()
+     * @param initDirectExchange : 需要绑定的交换机对象，参数名必须和某个@Bean的方法名完全相同，这样就会进行自动注入，对应 .to()
+     *                           .with() 方法对应的RoutingKey
+     * @return
+     */
+    @Bean
+    public Binding initBinding(DirectExchange initDirectExchange, Queue initFromServerQueue) {
+        return BindingBuilder.bind(initFromServerQueue).to(initDirectExchange).with(INIT_FROM_SERVER_KEY);
+    }
+
+}

server/src/main/java/io/wdd/rpc/message/ToAgentOrder.java

@@ -0,0 +1,55 @@
+package io.wdd.rpc.message;
+
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.wdd.common.beans.rabbitmq.OctopusMessage;
+import io.wdd.common.beans.rabbitmq.OctopusMessageType;
+import io.wdd.common.handler.MyRuntimeException;
+import io.wdd.rpc.init.FromServerMessageBinding;
+import lombok.SneakyThrows;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.actuate.amqp.RabbitHealthIndicator;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+/**
+ * adaptor
+ *  provide override method to convert Object and send to rabbitmq
+ */
+@Component
+public class ToAgentOrder {
+
+    @Resource
+    RabbitTemplate rabbitTemplate;
+
+    @Resource
+    FromServerMessageBinding fromServerMessageBinding;
+
+    /**
+     *
+     *  send to Queue -- InitFromServer
+     *
+     * @param message octopus message
+     */
+    public void send(OctopusMessage message){
+
+        // only accept INIT type message
+        if (!OctopusMessageType.INIT.equals(message.getType())) {
+            throw new MyRuntimeException("To Agent Order method usage wrong !");
+        }
+
+        // send to Queue -- InitFromServer
+
+        rabbitTemplate.convertAndSend(fromServerMessageBinding.INIT_EXCHANGE, fromServerMessageBinding.INIT_FROM_SERVER_KEY, writeData(message));
+
+    }
+
+    @SneakyThrows
+    private byte[] writeData(Object data){
+        ObjectMapper objectMapper = new ObjectMapper();
+        return objectMapper.writeValueAsBytes(data);
+    }
+
+}

server/src/main/java/io/wdd/server/controller/AppController.java

@@ -3,7 +3,7 @@ package io.wdd.server.controller;
 
 import io.wdd.server.beans.vo.AppInfoVO;
 import io.wdd.server.coreService.CoreAppService;
-import io.wdd.common.beans.R;
+import io.wdd.common.beans.response.R;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 

server/src/main/java/io/wdd/server/controller/DomainController.java

@@ -3,7 +3,7 @@ package io.wdd.server.controller;
 import io.wdd.server.beans.po.DomainInfoPO;
 import io.wdd.server.beans.vo.DomainInfoVO;
 import io.wdd.server.coreService.CoreDomainService;
-import io.wdd.common.beans.R;
+import io.wdd.common.beans.response.R;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 

server/src/main/java/io/wdd/server/controller/ServerController.java

@@ -7,7 +7,7 @@ import io.wdd.server.beans.vo.AppInfoVO;
 import io.wdd.server.beans.vo.DomainInfoVO;
 import io.wdd.server.beans.vo.ServerInfoVO;
 import io.wdd.server.coreService.CoreServerService;
-import io.wdd.common.beans.R;
+import io.wdd.common.beans.response.R;
 import org.apache.commons.lang3.ObjectUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.lang.Nullable;

server/src/main/java/io/wdd/server/utils/DaemonDatabaseOperator.java

@@ -0,0 +1,52 @@
+package io.wdd.server.utils;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.wdd.server.beans.vo.ServerInfoVO;
+import io.wdd.server.coreService.CoreServerService;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+/**
+ * The type Daemon database operator.
+ */
+@Component
+public class DaemonDatabaseOperator {
+
+    /**
+     * The Core server service.
+     */
+    @Resource
+    CoreServerService coreServerService;
+
+    private ThreadFactory threadFactory;
+
+    private ExecutorService fixedThreadPool;
+
+    /**
+     * Save init octopus agent info boolean.
+     *
+     * @param serverInfoVO the server info vo
+     * @return the result
+     */
+    public boolean saveInitOctopusAgentInfo(ServerInfoVO serverInfoVO) {
+
+        return coreServerService.serverCreate(serverInfoVO);
+    }
+
+    @PostConstruct
+    private void buildDaemonDatabaseThread() {
+
+        threadFactory = new ThreadFactoryBuilder()
+                .setDaemon(true)
+                .setNameFormat("database-daemon")
+                .setPriority(10)
+                .build();
+
+        fixedThreadPool = Executors.newFixedThreadPool(1, threadFactory);
+    }
+}

server/src/main/resources/application.yml

@@ -1,6 +1,21 @@
 server:
   port: 9999
 
+octopus:
+  message:
+    # agent boot up default common exchange
+    init_exchange: InitExchange
+    # server will send message to agent using this common queue
+    init_to_server: InitToServer
+    # agent boot up default common exchange routing key
+    init_to_server_key: InitToServerKey
+    # server will receive message from agent using this common queue
+    init_from_server: InitFromServer
+    # agent boot up default common exchange routing key
+    init_from_server_key: InitFromServerKey
+    # initialization register time out (unit ms) default is 5 min
+    init_ttl: "300000"
+
 spring:
   rabbitmq:
     host: 127.0.0.1