// Path: src/main/java/kafka/utils/SystemTime.java
// public class SystemTime implements Time {
//     public static final SystemTime instance = new SystemTime();
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
// 
//         }
//     }
// }


import kafka.common.KafkaException;
import kafka.utils.SystemTime;
import kafka.utils.Time;
import kafka.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.SocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
package kafka.network;



/**
 * Thread that processes all requests from a single connection. There are N of these running in parallel
 * each of which has its own selectors
 */
public class Processor extends AbstractServerThread {
    public int id;
    public Time time;
    public int maxRequestSize;
    public RequestChannel requestChannel;

    public Processor(int id, Time time, int maxRequestSize, RequestChannel requestChannel) {
        this.id = id;
        this.time = time;
        this.maxRequestSize = maxRequestSize;
        this.requestChannel = requestChannel;
    }

    private ConcurrentLinkedQueue<SocketChannel> newConnections = new ConcurrentLinkedQueue<SocketChannel>();

    Logger logger = LoggerFactory.getLogger(Processor.class);

    @Override
    public void run() {
        startupComplete();
        while (isRunning()) {
            // setup any new connections that have been queued up
            configureNewConnections();
            // register any new responses for writing
            processNewResponses();
// Path: src/main/java/kafka/utils/SystemTime.java
// public class SystemTime implements Time {
//     public static final SystemTime instance = new SystemTime();
// 
//     @Override
//     public long milliseconds() {
//         return System.currentTimeMillis();
//     }
// 
//     @Override
//     public long nanoseconds() {
//         return System.nanoTime();
//     }
// 
//     @Override
//     public void sleep(long ms) {
//         try {
//             Thread.sleep(ms);
//         } catch (InterruptedException e) {
// 
//         }
//     }
// }


// Path: src/main/java/kafka/network/Processor.java
import kafka.common.KafkaException;
import kafka.utils.SystemTime;
import kafka.utils.Time;
import kafka.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.SocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

package kafka.network;



/**
 * Thread that processes all requests from a single connection. There are N of these running in parallel
 * each of which has its own selectors
 */
public class Processor extends AbstractServerThread {
    public int id;
    public Time time;
    public int maxRequestSize;
    public RequestChannel requestChannel;

    public Processor(int id, Time time, int maxRequestSize, RequestChannel requestChannel) {
        this.id = id;
        this.time = time;
        this.maxRequestSize = maxRequestSize;
        this.requestChannel = requestChannel;
    }

    private ConcurrentLinkedQueue<SocketChannel> newConnections = new ConcurrentLinkedQueue<SocketChannel>();

    Logger logger = LoggerFactory.getLogger(Processor.class);

    @Override
    public void run() {
        startupComplete();
        while (isRunning()) {
            // setup any new connections that have been queued up
            configureNewConnections();
            // register any new responses for writing
            processNewResponses();
            long startSelectTime = SystemTime.instance.milliseconds();