<commit_before>package com.rvprg.raft.transport.impl;

import java.io.File;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage.MessageType;
import com.rvprg.raft.transport.ChannelPipelineInitializer;
import com.rvprg.raft.transport.MemberId;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.stream.ChunkedFile;
import io.netty.handler.stream.ChunkedWriteHandler;

public class SnapshotSender {
    private final Logger logger = LoggerFactory.getLogger(SnapshotSender.class);

    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;
    private final ServerBootstrap server;
    private final File fileName;
    private final String snapshotId;

    private final BiConsumer<MemberId, Channel> transferStart;
    private final Consumer<MemberId> transferCompleted;
    private final BiConsumer<MemberId, Throwable> transferCompletedExceptionally;

    private final ChannelPipelineInitializer channelPipelineInitializer;

    private class SnapshotTransferInitiator extends SimpleChannelInboundHandler<RaftMessage> {
        private MemberId memberId;
        public static final String SnapshotTransferInitiator = "SnapshotTransferInitiator";
        private static final String ChunkedWriteHandler = "ChunkedWriteHandler";

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            transferCompletedExceptionally.accept(memberId, cause);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RaftMessage msg) throws Exception {
            if (msg.getType() != MessageType.SnapshotDownloadRequest) {
                logger.error("Wrong message type. Ignoring.");
                return;
            }

            String memberIdStr = msg.getSnapshotDownloadRequest().getMemberId();
            try {
                memberId = MemberId.fromString(memberIdStr);
            } catch (IllegalArgumentException e) {
                logger.error("Could not parse memberId. MemberId string = {}.", memberIdStr, e);
                return;
            }

            String snapshotId = msg.getSnapshotDownloadRequest().getSnapshotId();
            if (!SnapshotSender.this.snapshotId.equalsIgnoreCase(snapshotId)) {
                ctx.channel().close();
                logger.info("MemberId={}. Requested snapshotId={}, but we are serving snapshotId={}. Closing connection.", memberIdStr, snapshotId, SnapshotSender.this.snapshotId);
                return;
            }

            for (String handlerName : channelPipelineInitializer.getHandlerNames()) {
                ctx.pipeline().remove(handlerName);
            }
            ctx.pipeline().addBefore(SnapshotTransferInitiator, ChunkedWriteHandler, new ChunkedWriteHandler());

            final ChunkedFile chunkedFile = new ChunkedFile(fileName);
            ctx.channel().writeAndFlush(chunkedFile).addListener(ChannelFutureListener.CLOSE).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    chunkedFile.close();
                    transferCompleted.accept(memberId);
                }
            });

            transferStart.accept(memberId, ctx.channel());
        }
    }

    public SnapshotSender(ChannelPipelineInitializer channelPipelineInitializer,
            MemberId memberId, String snapshotId, File fileName,
            BiConsumer<MemberId, Channel> transferStart,
            Consumer<MemberId> transferCompleted,
            BiConsumer<MemberId, Throwable> transferCompletedExceptionally) throws InterruptedException {
        this.bossGroup = new NioEventLoopGroup();
        this.workerGroup = new NioEventLoopGroup();
        this.server = new ServerBootstrap();
        this.fileName = fileName;
        this.snapshotId = snapshotId;<commit_msg>Consistent behavior for unexpected requests.<commit_after>package com.rvprg.raft.transport.impl;

import java.io.File;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage.MessageType;
import com.rvprg.raft.transport.ChannelPipelineInitializer;
import com.rvprg.raft.transport.MemberId;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.stream.ChunkedFile;
import io.netty.handler.stream.ChunkedWriteHandler;

public class SnapshotSender {
    private final Logger logger = LoggerFactory.getLogger(SnapshotSender.class);

    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;
    private final ServerBootstrap server;
    private final File fileName;
    private final String snapshotId;

    private final BiConsumer<MemberId, Channel> transferStart;
    private final Consumer<MemberId> transferCompleted;
    private final BiConsumer<MemberId, Throwable> transferCompletedExceptionally;

    private final ChannelPipelineInitializer channelPipelineInitializer;

    private class SnapshotTransferInitiator extends SimpleChannelInboundHandler<RaftMessage> {
        private MemberId memberId;
        public static final String SnapshotTransferInitiator = "SnapshotTransferInitiator";
        private static final String ChunkedWriteHandler = "ChunkedWriteHandler";

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            transferCompletedExceptionally.accept(memberId, cause);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RaftMessage msg) throws Exception {
            if (msg.getType() != MessageType.SnapshotDownloadRequest) {
                logger.error("Wrong message type. Closing connection.");
                ctx.channel().close();
                return;
            }

            String memberIdStr = msg.getSnapshotDownloadRequest().getMemberId();
            try {
                memberId = MemberId.fromString(memberIdStr);
            } catch (IllegalArgumentException e) {
                logger.error("Could not parse memberId. MemberId string = {}. Closing connection.", memberIdStr, e);
                ctx.channel().close();
                return;
            }

            String snapshotId = msg.getSnapshotDownloadRequest().getSnapshotId();
            if (!SnapshotSender.this.snapshotId.equalsIgnoreCase(snapshotId)) {
                logger.info("MemberId={}. Requested snapshotId={}, but we are serving snapshotId={}. Closing connection.", memberIdStr, snapshotId, SnapshotSender.this.snapshotId);
                ctx.channel().close();
                return;
            }

            for (String handlerName : channelPipelineInitializer.getHandlerNames()) {
                ctx.pipeline().remove(handlerName);
            }
            ctx.pipeline().addBefore(SnapshotTransferInitiator, ChunkedWriteHandler, new ChunkedWriteHandler());

            final ChunkedFile chunkedFile = new ChunkedFile(fileName);
            ctx.channel().writeAndFlush(chunkedFile).addListener(ChannelFutureListener.CLOSE).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    chunkedFile.close();
                    transferCompleted.accept(memberId);
                }
            });

            transferStart.accept(memberId, ctx.channel());
        }
    }

    public SnapshotSender(ChannelPipelineInitializer channelPipelineInitializer,
            MemberId memberId, String snapshotId, File fileName,
            BiConsumer<MemberId, Channel> transferStart,
            Consumer<MemberId> transferCompleted,
            BiConsumer<MemberId, Throwable> transferCompletedExceptionally) throws InterruptedException {
        this.bossGroup = new NioEventLoopGroup();
        this.workerGroup = new NioEventLoopGroup();
        this.server = new ServerBootstrap();
        this.fileName = fileName;
        this.snapshotId = snapshotId;
{'fraction_non_alphanumeric': 0.08120428283684936, 'fraction_numerical': 0.000636064878617619, 'mean_word_length': 3.241906474820144, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3314892', 'n_tokens_mistral': 2358, 'n_tokens_neox': 2309, 'n_words': 486}