<commit_before>package org.yamcs.tctm;

import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.yamcs.ConfigurationException;
import org.yamcs.YConfiguration;
import org.yamcs.YamcsServer;
import org.yamcs.cmdhistory.CommandHistoryPublisher;
import org.yamcs.commanding.PreparedCommand;
import org.yamcs.parameter.ParameterValue;
import org.yamcs.parameter.SystemParametersCollector;
import org.yamcs.parameter.SystemParametersProducer;
import org.yamcs.time.TimeService;
import org.yamcs.utils.LoggingUtils;
import org.yamcs.utils.TimeEncoding;
import org.yamcs.utils.YObjectLoader;

import com.google.common.util.concurrent.AbstractService;
import com.google.common.util.concurrent.RateLimiter;

/**
 * Sends raw packets on Tcp socket.
 * 
 * @author nm
 *
 */
public class TcpTcDataLink extends AbstractService implements Runnable, TcDataLink, SystemParametersProducer {

    protected SocketChannel socketChannel = null;
    protected String host = "whirl";
    protected int port = 10003;
    protected CommandHistoryPublisher commandHistoryListener;
    protected Selector selector;
    SelectionKey selectionKey;

    protected ScheduledThreadPoolExecutor timer;
    protected volatile boolean disabled = false;

    protected BlockingQueue<PreparedCommand> commandQueue;
    RateLimiter rateLimiter;

    protected volatile long tcCount;

    private String sv_linkStatus_id, sp_dataCount_id;

    private SystemParametersCollector sysParamCollector;
    protected final Logger log;
    private final String yamcsInstance;
    private final String name;
    TimeService timeService;
    static final PreparedCommand SIGNAL_QUIT = new PreparedCommand(new byte[0]);
    TcDequeueAndSend tcSender;

    CommandPostprocessor cmdPostProcessor;
    final YConfiguration config;

    public TcpTcDataLink(String yamcsInstance, String name, YConfiguration config) throws ConfigurationException {
        log = LoggingUtils.getLogger(this.getClass(), yamcsInstance);
        this.yamcsInstance = yamcsInstance;
        this.name = name;
        this.config = config;
        configure(yamcsInstance, config);
        timeService = YamcsServer.getTimeService(yamcsInstance);
    }

    public TcpTcDataLink(String yamcsInstance, String name, String spec) throws ConfigurationException {
        this(yamcsInstance, name, YConfiguration.getConfiguration("tcp").getConfig(spec));
    }

    private void configure(String yamcsInstance, YConfiguration config) {
        if (config.containsKey("tcHost")) {// this is when the config is specified in tcp.yaml
            host = config.getString("tcHost");
            port = config.getInt("tcPort");
        } else {
            host = config.getString("host");
            port = config.getInt("port");
        }
        initPostprocessor(yamcsInstance, config);

        if (config.containsKey("tcQueueSize")) {
            commandQueue = new LinkedBlockingQueue<>(config.getInt("tcQueueSize"));
        } else {
            commandQueue = new LinkedBlockingQueue<>();
        }
        if (config.containsKey("tcMaxRate")) {
            rateLimiter = RateLimiter.create(config.getInt("tcMaxRate"));
        }

    }

    protected long getCurrentTime() {
        if (timeService != null) {
            return timeService.getMissionTime();
        } else {
            return TimeEncoding.getWallclockTime();
        }
    }

    @Override
    protected void doStart() {
        setupSysVariables();
        this.timer = new ScheduledThreadPoolExecutor(2);
        openSocket();
        tcSender = new TcDequeueAndSend();
        timer.execute(tcSender);
        timer.scheduleAtFixedRate(this, 10L, 10L, TimeUnit.SECONDS);
        notifyStarted();
    }

    protected void initPostprocessor(String instance, YConfiguration config) {
        String commandPostprocessorClassName = IssCommandPostprocessor.class.getName();
        Object commandPostprocessorArgs = null;

        if (config != null) {
            commandPostprocessorClassName = config.getString("commandPostprocessorClassName",
                    IssCommandPostprocessor.class.getName());
            if (config.containsKey("commandPostprocessorArgs")) {
                commandPostprocessorArgs = config.getMap("commandPostprocessorArgs");
            }
        }

        try {
            if (commandPostprocessorArgs != null) {
                cmdPostProcessor = YObjectLoader.loadObject(commandPostprocessorClassName, instance,
                        commandPostprocessorArgs);<commit_msg>Prevent timeout on starting yamcs with tcp/tc link<commit_after>package org.yamcs.tctm;

import java.io.IOException;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.yamcs.ConfigurationException;
import org.yamcs.YConfiguration;
import org.yamcs.YamcsServer;
import org.yamcs.cmdhistory.CommandHistoryPublisher;
import org.yamcs.commanding.PreparedCommand;
import org.yamcs.parameter.ParameterValue;
import org.yamcs.parameter.SystemParametersCollector;
import org.yamcs.parameter.SystemParametersProducer;
import org.yamcs.time.TimeService;
import org.yamcs.utils.LoggingUtils;
import org.yamcs.utils.TimeEncoding;
import org.yamcs.utils.YObjectLoader;

import com.google.common.util.concurrent.AbstractService;
import com.google.common.util.concurrent.RateLimiter;

/**
 * Sends raw packets on Tcp socket.
 * 
 * @author nm
 *
 */
public class TcpTcDataLink extends AbstractService implements Runnable, TcDataLink, SystemParametersProducer {

    protected SocketChannel socketChannel = null;
    protected String host = "whirl";
    protected int port = 10003;
    protected CommandHistoryPublisher commandHistoryListener;
    protected Selector selector;
    SelectionKey selectionKey;

    protected ScheduledThreadPoolExecutor timer;
    protected volatile boolean disabled = false;

    protected BlockingQueue<PreparedCommand> commandQueue;
    RateLimiter rateLimiter;

    protected volatile long tcCount;

    private String sv_linkStatus_id, sp_dataCount_id;

    private SystemParametersCollector sysParamCollector;
    protected final Logger log;
    private final String yamcsInstance;
    private final String name;
    TimeService timeService;
    static final PreparedCommand SIGNAL_QUIT = new PreparedCommand(new byte[0]);
    TcDequeueAndSend tcSender;

    CommandPostprocessor cmdPostProcessor;
    final YConfiguration config;

    public TcpTcDataLink(String yamcsInstance, String name, YConfiguration config) throws ConfigurationException {
        log = LoggingUtils.getLogger(this.getClass(), yamcsInstance);
        this.yamcsInstance = yamcsInstance;
        this.name = name;
        this.config = config;
        configure(yamcsInstance, config);
        timeService = YamcsServer.getTimeService(yamcsInstance);
    }

    public TcpTcDataLink(String yamcsInstance, String name, String spec) throws ConfigurationException {
        this(yamcsInstance, name, YConfiguration.getConfiguration("tcp").getConfig(spec));
    }

    private void configure(String yamcsInstance, YConfiguration config) {
        if (config.containsKey("tcHost")) {// this is when the config is specified in tcp.yaml
            host = config.getString("tcHost");
            port = config.getInt("tcPort");
        } else {
            host = config.getString("host");
            port = config.getInt("port");
        }
        initPostprocessor(yamcsInstance, config);

        if (config.containsKey("tcQueueSize")) {
            commandQueue = new LinkedBlockingQueue<>(config.getInt("tcQueueSize"));
        } else {
            commandQueue = new LinkedBlockingQueue<>();
        }
        if (config.containsKey("tcMaxRate")) {
            rateLimiter = RateLimiter.create(config.getInt("tcMaxRate"));
        }

    }

    protected long getCurrentTime() {
        if (timeService != null) {
            return timeService.getMissionTime();
        } else {
            return TimeEncoding.getWallclockTime();
        }
    }

    @Override
    protected void doStart() {
        setupSysVariables();
        this.timer = new ScheduledThreadPoolExecutor(2);
        tcSender = new TcDequeueAndSend();
        timer.execute(tcSender);
        timer.scheduleAtFixedRate(this, 10L, 10L, TimeUnit.SECONDS);
        notifyStarted();
    }

    protected void initPostprocessor(String instance, YConfiguration config) {
        String commandPostprocessorClassName = IssCommandPostprocessor.class.getName();
        Object commandPostprocessorArgs = null;

        if (config != null) {
            commandPostprocessorClassName = config.getString("commandPostprocessorClassName",
                    IssCommandPostprocessor.class.getName());
            if (config.containsKey("commandPostprocessorArgs")) {
                commandPostprocessorArgs = config.getMap("commandPostprocessorArgs");
            }
        }

        try {
            if (commandPostprocessorArgs != null) {
                cmdPostProcessor = YObjectLoader.loadObject(commandPostprocessorClassName, instance,
                        commandPostprocessorArgs);
{'fraction_non_alphanumeric': 0.08528912413861979, 'fraction_numerical': 0.0023968840507340456, 'mean_word_length': 3.527124773960217, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '6269766', 'n_tokens_mistral': 2692, 'n_tokens_neox': 2619, 'n_words': 571}