class TxPullRequestHandler extends SimpleChannelInboundHandler<TxPullRequest> {

    @Override
    protected void channelRead0( ChannelHandlerContext ctx, final TxPullRequest msg ) throws Exception
    {
        monitor.increment();

        if ( msg.previousTxId() <= 0 )
        {
            log.error( "Illegal tx pull request" );
            endInteraction( ctx, E_INVALID_REQUEST, -1 );
            return;
        }

        StoreId localStoreId = storeIdSupplier.get();
        StoreId expectedStoreId = msg.expectedStoreId();

        long firstTxId = msg.previousTxId() + 1;

        /*
         * This is the minimum transaction id we must send to consider our streaming operation successful. The kernel can
         * concurrently prune even future transactions while iterating and the cursor will silently fail on iteration, so
         * we need to add our own protection for this reason and also as a generally important sanity check for the fulfillment
         * of the consistent recovery contract which requires us to stream transactions at least as far as the time when the
         * file copy operation completed.
         */
        long txIdPromise = transactionIdStore.getLastCommittedTransactionId();
        IOCursor<CommittedTransactionRepresentation> txCursor = getCursor( txIdPromise, ctx, firstTxId, localStoreId, expectedStoreId );

        if ( txCursor != null )
        {
            ChunkedTransactionStream txStream = new ChunkedTransactionStream( log, localStoreId, firstTxId, txIdPromise, txCursor, protocol );
            // chunked transaction stream ends the interaction internally and closes the cursor
            ctx.writeAndFlush( txStream ).addListener( f ->
            {
                if ( log.isDebugEnabled() || !f.isSuccess() )
                {
                    String message = format( "Streamed transactions [%d--%d] to %s", firstTxId, txStream.lastTxId(), ctx.channel().remoteAddress() );
                    if ( f.isSuccess() )
                    {
                        log.debug( message );
                    }
                    else
                    {
                        log.warn( message, f.cause() );
                    }
                }
            } );
        }
    }

    public  TxPullRequestHandler( CatchupServerProtocol protocol, Supplier<StoreId> storeIdSupplier,
            BooleanSupplier databaseAvailable, Supplier<NeoStoreDataSource> dataSourceSupplier, Monitors monitors, LogProvider logProvider );

}