/*
 * 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.
 */

package org.apache.skywalking.apm.agent.core.commands;

import org.apache.skywalking.apm.agent.core.boot.BootService;
import org.apache.skywalking.apm.agent.core.boot.DefaultImplementor;
import org.apache.skywalking.apm.agent.core.boot.ServiceManager;
import org.apache.skywalking.apm.agent.core.logging.api.ILog;
import org.apache.skywalking.apm.agent.core.logging.api.LogManager;
import org.apache.skywalking.apm.network.common.Command;
import org.apache.skywalking.apm.network.common.Commands;
import org.apache.skywalking.apm.network.trace.component.command.BaseCommand;
import org.apache.skywalking.apm.network.trace.component.command.CommandDeserializer;
import org.apache.skywalking.apm.network.trace.component.command.UnsupportedCommandException;
import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;

import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

@DefaultImplementor
public class CommandService implements BootService, Runnable {

    private static final ILog LOGGER = LogManager.getLogger(CommandService.class);

    private volatile boolean isRunning = true;
    private ExecutorService executorService = Executors.newSingleThreadExecutor();
    private LinkedBlockingQueue<BaseCommand> commands = new LinkedBlockingQueue<BaseCommand>(64);
    private CommandSerialNumberCache serialNumberCache = new CommandSerialNumberCache();

    @Override
    public void prepare() throws Throwable {
    }

    @Override
    public void boot() throws Throwable {
        executorService.submit(new RunnableWithExceptionProtection(this, new RunnableWithExceptionProtection.CallbackWhenException() {
            @Override
            public void handle(final Throwable t) {
                LOGGER.error(t, "CommandService failed to execute commands");
            }
        }));
    }

    @Override
    public void run() {
        final CommandExecutorService commandExecutorService = ServiceManager.INSTANCE.findService(CommandExecutorService.class);

        while (isRunning) {
            try {
                BaseCommand command = commands.take();

                if (isCommandExecuted(command)) {
                    continue;
                }

                commandExecutorService.execute(command);
                serialNumberCache.add(command.getSerialNumber());
            } catch (InterruptedException e) {
                LOGGER.error(e, "Failed to take commands.");
            } catch (CommandExecutionException e) {
                LOGGER.error(e, "Failed to execute command[{}].", e.command().getCommand());
            } catch (Throwable e) {
                LOGGER.error(e, "There is unexpected exception");
            }
        }
    }

    private boolean isCommandExecuted(BaseCommand command) {
        return serialNumberCache.contain(command.getSerialNumber());
    }

    @Override
    public void onComplete() throws Throwable {

    }

    @Override
    public void shutdown() throws Throwable {
        isRunning = false;
        commands.drainTo(new ArrayList<BaseCommand>());
        executorService.shutdown();
    }

    public void receiveCommand(Commands commands) {
        for (Command command : commands.getCommandsList()) {
            try {
                BaseCommand baseCommand = CommandDeserializer.deserialize(command);

                if (isCommandExecuted(baseCommand)) {
                    LOGGER.warn("Command[{}] is executed, ignored", baseCommand.getCommand());
                    continue;
                }

                boolean success = this.commands.offer(baseCommand);

                if (!success && LOGGER.isWarnEnable()) {
                    LOGGER.warn("Command[{}, {}] cannot add to command list. because the command list is full.", baseCommand
                        .getCommand(), baseCommand.getSerialNumber());
                }
            } catch (UnsupportedCommandException e) {
                if (LOGGER.isWarnEnable()) {
                    LOGGER.warn("Received unsupported command[{}].", e.getCommand().getCommand());
                }
            }
        }
    }
}

@@ -855,16 +855,177 @@
 mands;%0A%0A
+import java.util.ArrayList;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors;%0Aimport java.util.concurrent.LinkedBlockingQueue;%0A
 import o
@@ -1147,16 +1147,92 @@
 mentor;%0A
+import org.apache.skywalking.apm.agent.core.boot.DefaultNamedThreadFactory;%0A
 import o
@@ -1867,170 +1867,8 @@
 n;%0A%0A
-import java.util.ArrayList;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors;%0Aimport java.util.concurrent.LinkedBlockingQueue;%0A%0A
 @Def
@@ -2158,16 +2158,77 @@
 xecutor(
+%0A        new DefaultNamedThreadFactory(%22CommandService%22)%0A    
 );%0A    p
@@ -2302,27 +2302,16 @@
 ngQueue%3C
-BaseCommand
 %3E(64);%0A 
@@ -2552,16 +2552,29 @@
 .submit(
+%0A            
 new Runn
@@ -2611,159 +2611,12 @@
 is, 
-new RunnableWithExceptionProtection.CallbackWhenException() %7B%0A            @Override%0A            public void handle(final Throwable t) %7B%0A               
+t -%3E
  LOG
@@ -2672,31 +2672,17 @@
 mmands%22)
-;%0A            %7D
+)
 %0A       
@@ -2686,10 +2686,8 @@
     
-%7D)
 );%0A 
@@ -3903,27 +3903,16 @@
 rayList%3C
-BaseCommand
 %3E());%0A  
@@ -4516,32 +4516,57 @@
     LOGGER.warn(
+%0A                        
 %22Command%5B%7B%7D, %7B%7D%5D
@@ -4629,28 +4629,16 @@
  full.%22,
- baseCommand
 %0A       
@@ -4654,16 +4654,27 @@
         
+baseCommand
 .getComm
@@ -4701,32 +4701,53 @@
 etSerialNumber()
+%0A                    
 );%0A             

Make `CommandService` daemon to avoid blocking target application shutting down gracefully (#4541)