<commit_before>package org.arabellan.mud.network;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.Set;

import static org.arabellan.mud.network.Socket.State.CLOSED;

@Slf4j
public class SocketProcessor implements Runnable {

    private final Queue<Socket> socketQueue;
    private final Map<Integer, Socket> socketMap;
    private final EventBus eventBus;

    private Selector readSelector;
    private Selector writeSelector;

    public SocketProcessor(Queue<Socket> socketQueue, EventBus eventBus) {
        this.socketQueue = socketQueue;
        this.socketMap = new HashMap<>();
        this.eventBus = eventBus;
        this.eventBus.register(new OutgoingMessageHandler());
        this.eventBus.register(new BroadcastMessageHandler());
    }

    public void run() {
        try {
            this.readSelector = Selector.open();
            this.writeSelector = Selector.open();

            while (true) {
                cullClosedSockets();
                acceptNewSockets();
                readSockets();
                writeSockets();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void cullClosedSockets() {
        socketMap.entrySet().removeIf(entry -> {
            boolean isClosed = entry.getValue().getState() == CLOSED;
            if (isClosed) log.info("Socket closed: " + entry.getValue().getId());
            return isClosed;
        });
    }

    private void acceptNewSockets() {
        Socket socket = socketQueue.poll();
        while (socket != null) {
            log.info("Socket accepted: " + socket.getId());
            socket.setNonBlockingMode();
            socket.addReadSelector(readSelector);
            socketMap.put(socket.getId(), socket);
            socket = socketQueue.poll();
        }
    }

    private void readSockets() {
        try {
            int keysReady = readSelector.selectNow();
            if (keysReady > 0) {
                Set<SelectionKey> keys = readSelector.selectedKeys();
                for (SelectionKey key : keys) {
                    Socket socket = (Socket) key.attachment();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);

                    log.debug("Socket " + socket.getId() + " read selected");

                    int bytesRead = socket.getSocketChannel().read(buffer);
                    while (bytesRead > 0) {
                        bytesRead = socket.getSocketChannel().read(buffer);
                    }

                    if (bytesRead == -1) {
                        key.attach(null);
                        key.cancel();
                        socket.setState(CLOSED);
                    }

                    buffer.flip();

                    while (!buffer.hasRemaining()) {
                        if (TelnetProtocol.nextByteIAC(buffer)) {
                            ByteBuffer request = TelnetProtocol.extractTelnetCommand(buffer);
                            ByteBuffer response = TelnetProtocol.parseTelnetCommand(request);
                            socket.addWriteSelector(writeSelector);
                            socket.getOutgoingQueue().add(response);
                        } else {
                            byte[] bytes = new byte[buffer.limit()];
                            buffer.get(bytes);
                            String output = new String(bytes);
                            String trimmedOutput = output.trim();
                            log.debug("From " + socket.getId() + ": " + trimmedOutput);
                            trimmedOutput += "\r\n";
                            ByteBuffer outputBuffer = ByteBuffer.allocate(trimmedOutput.length());
                            outputBuffer.put(trimmedOutput.getBytes());
                            IncomingMessageEvent message = new IncomingMessageEvent(socket.getId(), outputBuffer);
                            eventBus.post(message);
                        }
                    }
                }

                keys.clear();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void writeSockets() {
        try {
            int keysReady = writeSelector.selectNow();
            if (keysReady > 0) {
                Set<SelectionKey> keys = writeSelector.selectedKeys();

                for (SelectionKey key : keys) {
                    Socket socket = (Socket) key.attachment();
                    ByteBuffer message = socket.getOutgoingQueue().poll();

                    if (message != null && message.hasRemaining()) {

                        if (socket.getProtocol() == Socket.Protocol.TELNET && !TelnetProtocol.nextByteIAC(message)) {
                            log.debug("To " + socket.getId() + ": " + StringUtils.fromByteBuffer(message));
                        }

                        socket.getSocketChannel().write(message);
                    }

                    if (socket.getOutgoingQueue().size() == 0) {
                        socket.removeWriteSelector();
                    }
                }

                keys.clear();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private class OutgoingMessageHandler {
        @Subscribe
        public void handle(OutgoingMessageEvent event) {
            log.trace("Handling OutgoingMessageEvent");
            Socket socket = socketMap.get(event.getId());
            socket.addWriteSelector(writeSelector);
            socket.getOutgoingQueue().add(event.getBuffer());
        }
    }
<commit_msg>Accidentally inverted some conditionals with the last commit.<commit_after>package org.arabellan.mud.network;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.HashMap;
import java.util.Map;
import java.util.Queue;
import java.util.Set;

import static org.arabellan.mud.network.Socket.State.CLOSED;

@Slf4j
public class SocketProcessor implements Runnable {

    private final Queue<Socket> socketQueue;
    private final Map<Integer, Socket> socketMap;
    private final EventBus eventBus;

    private Selector readSelector;
    private Selector writeSelector;

    public SocketProcessor(Queue<Socket> socketQueue, EventBus eventBus) {
        this.socketQueue = socketQueue;
        this.socketMap = new HashMap<>();
        this.eventBus = eventBus;
        this.eventBus.register(new OutgoingMessageHandler());
        this.eventBus.register(new BroadcastMessageHandler());
    }

    public void run() {
        try {
            this.readSelector = Selector.open();
            this.writeSelector = Selector.open();

            while (true) {
                cullClosedSockets();
                acceptNewSockets();
                readSockets();
                writeSockets();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void cullClosedSockets() {
        socketMap.entrySet().removeIf(entry -> {
            boolean isClosed = entry.getValue().getState() == CLOSED;
            if (isClosed) log.info("Socket closed: " + entry.getValue().getId());
            return isClosed;
        });
    }

    private void acceptNewSockets() {
        Socket socket = socketQueue.poll();
        while (socket != null) {
            log.info("Socket accepted: " + socket.getId());
            socket.setNonBlockingMode();
            socket.addReadSelector(readSelector);
            socketMap.put(socket.getId(), socket);
            socket = socketQueue.poll();
        }
    }

    private void readSockets() {
        try {
            int keysReady = readSelector.selectNow();
            if (keysReady > 0) {
                Set<SelectionKey> keys = readSelector.selectedKeys();
                for (SelectionKey key : keys) {
                    Socket socket = (Socket) key.attachment();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);

                    log.debug("Socket " + socket.getId() + " read selected");

                    int bytesRead = socket.getSocketChannel().read(buffer);
                    while (bytesRead > 0) {
                        bytesRead = socket.getSocketChannel().read(buffer);
                    }

                    if (bytesRead == -1) {
                        key.attach(null);
                        key.cancel();
                        socket.setState(CLOSED);
                    }

                    buffer.flip();

                    while (buffer.hasRemaining()) {
                        if (TelnetProtocol.nextByteIAC(buffer)) {
                            ByteBuffer request = TelnetProtocol.extractTelnetCommand(buffer);
                            ByteBuffer response = TelnetProtocol.parseTelnetCommand(request);
                            socket.addWriteSelector(writeSelector);
                            socket.getOutgoingQueue().add(response);
                        } else {
                            byte[] bytes = new byte[buffer.limit()];
                            buffer.get(bytes);
                            String output = new String(bytes);
                            String trimmedOutput = output.trim();
                            log.debug("From " + socket.getId() + ": " + trimmedOutput);
                            trimmedOutput += "\r\n";
                            ByteBuffer outputBuffer = ByteBuffer.allocate(trimmedOutput.length());
                            outputBuffer.put(trimmedOutput.getBytes());
                            outputBuffer.flip();
                            IncomingMessageEvent message = new IncomingMessageEvent(socket.getId(), outputBuffer);
                            eventBus.post(message);
                        }
                    }
                }

                keys.clear();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private void writeSockets() {
        try {
            int keysReady = writeSelector.selectNow();
            if (keysReady > 0) {
                Set<SelectionKey> keys = writeSelector.selectedKeys();

                for (SelectionKey key : keys) {
                    Socket socket = (Socket) key.attachment();
                    ByteBuffer message = socket.getOutgoingQueue().poll();

                    if (message != null && !message.hasRemaining()) {

                        if (socket.getProtocol() == Socket.Protocol.TELNET && !TelnetProtocol.nextByteIAC(message)) {
                            log.debug("To " + socket.getId() + ": " + StringUtils.fromByteBuffer(message));
                        }

                        socket.getSocketChannel().write(message);
                    }

                    if (socket.getOutgoingQueue().size() == 0) {
                        socket.removeWriteSelector();
                    }
                }

                keys.clear();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private class OutgoingMessageHandler {
        @Subscribe
        public void handle(OutgoingMessageEvent event) {
            log.trace("Handling OutgoingMessageEvent");
            Socket socket = socketMap.get(event.getId());
            socket.addWriteSelector(writeSelector);
            socket.getOutgoingQueue().add(event.getBuffer());
        }
    }

{'fraction_non_alphanumeric': 0.10180142438206954, 'fraction_numerical': 0.0020108923334729786, 'mean_word_length': 1.5575316048853654, 'pattern_counts': {'":': 4, '<': 15, '<?xml version=': 0, '>': 23, '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': '1461211', 'n_tokens_mistral': 2962, 'n_tokens_neox': 2800, 'n_words': 554}