<commit_before>/*
 * Copyright 2017 flow.ci
 *
 * Licensed 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 com.flow.platform.cc.service;

import static com.flow.platform.domain.CmdType.CREATE_SESSION;
import static com.flow.platform.domain.CmdType.DELETE_SESSION;
import static com.flow.platform.domain.CmdType.KILL;
import static com.flow.platform.domain.CmdType.RUN_SHELL;
import static com.flow.platform.domain.CmdType.SHUTDOWN;
import static com.flow.platform.domain.CmdType.STOP;

import com.flow.platform.cc.config.AppConfig;
import com.flow.platform.cc.dao.AgentDao;
import com.flow.platform.cc.dao.CmdDao;
import com.flow.platform.cc.dao.CmdResultDao;
import com.flow.platform.cc.domain.CmdQueueItem;
import com.flow.platform.cc.domain.CmdStatusItem;
import com.flow.platform.cc.exception.AgentErr;
import com.flow.platform.core.queue.PlatformQueue;
import com.flow.platform.core.service.WebhookServiceImplBase;
import com.flow.platform.core.task.WebhookCallBackTask;
import com.flow.platform.core.exception.IllegalStatusException;
import com.flow.platform.domain.Agent;
import com.flow.platform.domain.AgentPath;
import com.flow.platform.domain.AgentStatus;
import com.flow.platform.domain.Cmd;
import com.flow.platform.domain.CmdBase;
import com.flow.platform.domain.CmdInfo;
import com.flow.platform.domain.CmdResult;
import com.flow.platform.domain.CmdType;
import com.flow.platform.domain.Zone;
import com.flow.platform.core.exception.IllegalParameterException;
import com.flow.platform.util.Logger;
import com.flow.platform.util.zk.ZKClient;
import com.google.common.base.Strings;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.ZonedDateTime;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import javax.annotation.PostConstruct;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

/**
 * @author <EMAIL>
 */

@Service
@Transactional(isolation = Isolation.REPEATABLE_READ)
public class CmdServiceImpl extends WebhookServiceImplBase implements CmdService {

    private final static Logger LOGGER = new Logger(CmdService.class);

    @Value("${mq.queue.cmd.name}")
    private String cmdQueueName;

    @Autowired
    private AgentService agentService;

    @Autowired
    private ZoneService zoneService;

    @Autowired
    private PlatformQueue<CmdStatusItem> cmdStatusQueue;

    @Autowired
    private CmdDao cmdDao;

    @Autowired
    private CmdResultDao cmdResultDao;

    @Autowired
    private AgentDao agentDao;

    @Autowired
    private PlatformQueue<Message> cmdQueue;

    @Autowired
    protected ZKClient zkClient;

    private final Map<EnumSet<CmdType>, CmdValidator> cmdValidators = new HashMap<>();

    @PostConstruct
    public void init() {
        CmdValidatorForZoneRequired zoneRequired = new CmdValidatorForZoneRequired();
        cmdValidators.put(zoneRequired.targets(), zoneRequired);

        CmdValidatorForSessionRequired sessionRequired = new CmdValidatorForSessionRequired();
        cmdValidators.put(sessionRequired.targets(), sessionRequired);

        CmdValidatorForZoneAndNameRequired zoneAndNameRequired = new CmdValidatorForZoneAndNameRequired();
        cmdValidators.put(zoneAndNameRequired.targets(), zoneAndNameRequired);

        CmdValidatorForValidSession validSession = new CmdValidatorForValidSession();
        cmdValidators.put(validSession.targets(), validSession);
    }

    @Override
    @Transactional(noRollbackFor = Throwable.class)
    public Cmd create(CmdInfo info) {
        return create(info, 0);
    }

    @Override
    @Transactional(noRollbackFor = Throwable.class)
    public Cmd create(CmdInfo info, Integer retry) {
        Cmd cmd = Cmd.convert(info);
        cmd.setId(UUID.randomUUID().toString());
        cmd.setCreatedDate(ZonedDateTime.now());
        cmd.setUpdatedDate(ZonedDateTime.now());

        // check customized id from CmdInfo, therefore cmd id can be controlled by other system
        if (!Strings.isNullOrEmpty(info.getCustomizedId())) {
            cmd.setId(info.getCustomizedId());
        }

        // validate input cmd
        for (Map.Entry<EnumSet<CmdType>, CmdValidator> entry : cmdValidators.entrySet()) {
            EnumSet<CmdType> types = entry.getKey();
            CmdValidator validator = entry.getValue();

            if (types.contains(cmd.getType())) {
                validator.validate(cmd);
            }
        }

        // auto create session id when create cmd
        if (cmd.getType() == CmdType.CREATE_SESSION) {
            cmd.setSessionId(UUID.randomUUID().toString());
            LOGGER.traceMarker("create", "Create session id when cmd created: %s", cmd.getSessionId());
        }

        if (cmd.getTimeout() == null) {
            cmd.setTimeout(DEFAULT_CMD_TIMEOUT);
        }

        if (retry != null && retry > 0) {
            cmd.setRetry(retry);
        }

        return cmdDao.save(cmd);
    }

    @Override
    public void save(Cmd cmd) {
        cmdDao.update(cmd);
    }

    @Override
    public Cmd find(String cmdId) {
        return cmdDao.get(cmdId);
    }

    @Override
    public List<Cmd> listByAgentPath(AgentPath agentPath) {
        return cmdDao.list(agentPath, null, null);
    }

    @Override
    public List<Cmd> listBySession(String sessionId) {
        return cmdDao.list(sessionId);
    }

    @Override
    public List<CmdResult> listResult(Set<String> cmdIds) {
        return cmdResultDao.list(cmdIds);
    }

    @Override
    public List<Cmd> listWorkingCmd(AgentPath agentPath) {
        return cmdDao.list(agentPath, Sets.newHashSet(CmdType.RUN_SHELL), Cmd.WORKING_STATUS);
    }

    @Override
    @Transactional(propagation = Propagation.NEVER)
    public Cmd enqueue(CmdInfo cmdInfo, int priority, int retry) {
        Cmd cmd = create(cmdInfo, retry);

        CmdQueueItem item = new CmdQueueItem(cmd.getId(), priority, retry);
        MessageProperties properties = new MessageProperties();
        properties.setPriority(item.getPriority());
        cmdQueue.enqueue(new Message(item.toBytes(), properties));

        return cmd;
    }

    @Override
    public void updateStatus(CmdStatusItem statusItem, boolean inQueue) {
        if (inQueue) {
            LOGGER.trace("Report cmd status from queue: %s", statusItem.getCmdId());
            cmdStatusQueue.enqueue(statusItem);
        }

        LOGGER.trace("Report cmd %s to status %s", statusItem.getCmdId(), statusItem.getStatus());
        String cmdId = statusItem.getCmdId();
        Cmd cmd = find(cmdId);
        if (cmd == null) {
            throw new IllegalArgumentException("Cmd does not exist");
        }

        //TODO: missing unit test
        // set cmd status in sequence
        if (!cmd.addStatus(statusItem.getStatus())) {
            return;
        }

        // compare exiting cmd result and update
        CmdResult inputResult = statusItem.getCmdResult();

        if (inputResult != null) {
            inputResult.setCmdId(cmdId);
            cmdResultDao.saveOrUpdate(inputResult);
            cmd.setCmdResult(inputResult);<commit_msg>fix duplicate process CmdStatusItem for queue since missing return<commit_after>/*
 * Copyright 2017 flow.ci
 *
 * Licensed 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 com.flow.platform.cc.service;

import static com.flow.platform.domain.CmdType.CREATE_SESSION;
import static com.flow.platform.domain.CmdType.DELETE_SESSION;
import static com.flow.platform.domain.CmdType.KILL;
import static com.flow.platform.domain.CmdType.RUN_SHELL;
import static com.flow.platform.domain.CmdType.SHUTDOWN;
import static com.flow.platform.domain.CmdType.STOP;

import com.flow.platform.cc.config.AppConfig;
import com.flow.platform.cc.dao.AgentDao;
import com.flow.platform.cc.dao.CmdDao;
import com.flow.platform.cc.dao.CmdResultDao;
import com.flow.platform.cc.domain.CmdQueueItem;
import com.flow.platform.cc.domain.CmdStatusItem;
import com.flow.platform.cc.exception.AgentErr;
import com.flow.platform.core.queue.PlatformQueue;
import com.flow.platform.core.service.WebhookServiceImplBase;
import com.flow.platform.core.task.WebhookCallBackTask;
import com.flow.platform.core.exception.IllegalStatusException;
import com.flow.platform.domain.Agent;
import com.flow.platform.domain.AgentPath;
import com.flow.platform.domain.AgentStatus;
import com.flow.platform.domain.Cmd;
import com.flow.platform.domain.CmdBase;
import com.flow.platform.domain.CmdInfo;
import com.flow.platform.domain.CmdResult;
import com.flow.platform.domain.CmdType;
import com.flow.platform.domain.Zone;
import com.flow.platform.core.exception.IllegalParameterException;
import com.flow.platform.util.Logger;
import com.flow.platform.util.zk.ZKClient;
import com.google.common.base.Strings;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.ZonedDateTime;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import javax.annotation.PostConstruct;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

/**
 * @author <EMAIL>
 */

@Service
@Transactional(isolation = Isolation.REPEATABLE_READ)
public class CmdServiceImpl extends WebhookServiceImplBase implements CmdService {

    private final static Logger LOGGER = new Logger(CmdService.class);

    @Value("${mq.queue.cmd.name}")
    private String cmdQueueName;

    @Autowired
    private AgentService agentService;

    @Autowired
    private ZoneService zoneService;

    @Autowired
    private PlatformQueue<CmdStatusItem> cmdStatusQueue;

    @Autowired
    private CmdDao cmdDao;

    @Autowired
    private CmdResultDao cmdResultDao;

    @Autowired
    private AgentDao agentDao;

    @Autowired
    private PlatformQueue<Message> cmdQueue;

    @Autowired
    protected ZKClient zkClient;

    private final Map<EnumSet<CmdType>, CmdValidator> cmdValidators = new HashMap<>();

    @PostConstruct
    public void init() {
        CmdValidatorForZoneRequired zoneRequired = new CmdValidatorForZoneRequired();
        cmdValidators.put(zoneRequired.targets(), zoneRequired);

        CmdValidatorForSessionRequired sessionRequired = new CmdValidatorForSessionRequired();
        cmdValidators.put(sessionRequired.targets(), sessionRequired);

        CmdValidatorForZoneAndNameRequired zoneAndNameRequired = new CmdValidatorForZoneAndNameRequired();
        cmdValidators.put(zoneAndNameRequired.targets(), zoneAndNameRequired);

        CmdValidatorForValidSession validSession = new CmdValidatorForValidSession();
        cmdValidators.put(validSession.targets(), validSession);
    }

    @Override
    @Transactional(noRollbackFor = Throwable.class)
    public Cmd create(CmdInfo info) {
        return create(info, 0);
    }

    @Override
    @Transactional(noRollbackFor = Throwable.class)
    public Cmd create(CmdInfo info, Integer retry) {
        Cmd cmd = Cmd.convert(info);
        cmd.setId(UUID.randomUUID().toString());
        cmd.setCreatedDate(ZonedDateTime.now());
        cmd.setUpdatedDate(ZonedDateTime.now());

        // check customized id from CmdInfo, therefore cmd id can be controlled by other system
        if (!Strings.isNullOrEmpty(info.getCustomizedId())) {
            cmd.setId(info.getCustomizedId());
        }

        // validate input cmd
        for (Map.Entry<EnumSet<CmdType>, CmdValidator> entry : cmdValidators.entrySet()) {
            EnumSet<CmdType> types = entry.getKey();
            CmdValidator validator = entry.getValue();

            if (types.contains(cmd.getType())) {
                validator.validate(cmd);
            }
        }

        // auto create session id when create cmd
        if (cmd.getType() == CmdType.CREATE_SESSION) {
            cmd.setSessionId(UUID.randomUUID().toString());
            LOGGER.traceMarker("create", "Create session id when cmd created: %s", cmd.getSessionId());
        }

        if (cmd.getTimeout() == null) {
            cmd.setTimeout(DEFAULT_CMD_TIMEOUT);
        }

        if (retry != null && retry > 0) {
            cmd.setRetry(retry);
        }

        return cmdDao.save(cmd);
    }

    @Override
    public void save(Cmd cmd) {
        cmdDao.update(cmd);
    }

    @Override
    public Cmd find(String cmdId) {
        return cmdDao.get(cmdId);
    }

    @Override
    public List<Cmd> listByAgentPath(AgentPath agentPath) {
        return cmdDao.list(agentPath, null, null);
    }

    @Override
    public List<Cmd> listBySession(String sessionId) {
        return cmdDao.list(sessionId);
    }

    @Override
    public List<CmdResult> listResult(Set<String> cmdIds) {
        return cmdResultDao.list(cmdIds);
    }

    @Override
    public List<Cmd> listWorkingCmd(AgentPath agentPath) {
        return cmdDao.list(agentPath, Sets.newHashSet(CmdType.RUN_SHELL), Cmd.WORKING_STATUS);
    }

    @Override
    @Transactional(propagation = Propagation.NEVER)
    public Cmd enqueue(CmdInfo cmdInfo, int priority, int retry) {
        Cmd cmd = create(cmdInfo, retry);

        CmdQueueItem item = new CmdQueueItem(cmd.getId(), priority, retry);
        MessageProperties properties = new MessageProperties();
        properties.setPriority(item.getPriority());
        cmdQueue.enqueue(new Message(item.toBytes(), properties));

        return cmd;
    }

    @Override
    public void updateStatus(CmdStatusItem statusItem, boolean inQueue) {
        if (inQueue) {
            LOGGER.trace("Report cmd status from queue: %s", statusItem.getCmdId());
            cmdStatusQueue.enqueue(statusItem);
            return;
        }

        LOGGER.trace("Report cmd %s to status %s", statusItem.getCmdId(), statusItem.getStatus());
        String cmdId = statusItem.getCmdId();
        Cmd cmd = find(cmdId);
        if (cmd == null) {
            throw new IllegalArgumentException("Cmd does not exist");
        }

        //TODO: missing unit test
        // set cmd status in sequence
        if (!cmd.addStatus(statusItem.getStatus())) {
            return;
        }

        // compare exiting cmd result and update
        CmdResult inputResult = statusItem.getCmdResult();

        if (inputResult != null) {
            inputResult.setCmdId(cmdId);
            cmdResultDao.saveOrUpdate(inputResult);
            cmd.setCmdResult(inputResult);
{'fraction_non_alphanumeric': 0.10103117363056566, 'fraction_numerical': 0.0011921082434285034, 'mean_word_length': 3.505370569280344, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1468330', 'n_tokens_mistral': 4742, 'n_tokens_neox': 4729, 'n_words': 1056}