// Path: fast-ftrl-proximal/src/main/java/io/scaledml/core/BaseDisruptorRunner.java
// public abstract class BaseDisruptorRunner {
//     private Disruptor<? extends TwoPhaseEvent<?>> disruptor;
//     private Supplier<InputStream> inputStreamFactory;
//     private boolean skipFirst;
//     private Phaser phaser;
// 
//     public void process() throws IOException {
//         try (FastBufferedInputStream stream = new FastBufferedInputStream(inputStreamFactory.get())) {
//             Preconditions.checkArgument(phaser.getRegisteredParties() == 0);
//             phaser.register();
//             disruptor.start();
//             RingBuffer<? extends TwoPhaseEvent> ringBuffer = disruptor.getRingBuffer();
//             long cursor = ringBuffer.next();
//             LineBytesBuffer buffer = ringBuffer.get(cursor).input();
//             long lineNo = 0;
//             ringBuffer.get(cursor).lineNo(lineNo);
//             boolean needToSkipNext = skipFirst;
//             while (buffer.readLineFrom(stream)) {
//                 if (needToSkipNext) {
//                     needToSkipNext = false;
//                     continue;
//                 }
//                 lineNo++;
//                 ringBuffer.publish(cursor);
//                 cursor = ringBuffer.next();
//                 buffer = ringBuffer.get(cursor).input();
//                 ringBuffer.get(cursor).lineNo(lineNo);
//             }
//             disruptor.shutdown();
//             phaser.arriveAndAwaitAdvance();
//             phaser.arriveAndDeregister();
//         } finally {
//             afterDisruptorProcessed();
//         }
//     }
// 
//     protected abstract void afterDisruptorProcessed() throws IOException;
// 
//     protected void setDisruptor(Disruptor<? extends TwoPhaseEvent<?>> disruptor) {
//         this.disruptor = disruptor;
//     }
// 
//     @Inject
//     public void setInputStreamFactory(Supplier<InputStream> inputStreamFactory) {
//         this.inputStreamFactory = inputStreamFactory;
//     }
// 
//     @Inject
//     public void setSkipFirst(@Named("skipFirst") boolean skipFirst) {
//         this.skipFirst = skipFirst;
//     }
// 
//     @Inject
//     public void setPhaser(Phaser phaser) {
//         this.phaser = phaser;
//     }
// }
// 
// Path: fast-ftrl-proximal/src/main/java/io/scaledml/core/outputformats/OutputFormat.java
// public interface OutputFormat extends Closeable {
//     void emit(SparseItem item, double prediction);
// }
// 
// Path: fast-ftrl-proximal/src/main/java/io/scaledml/core/TwoPhaseEvent.java
// public class TwoPhaseEvent<T> {
//     public static <T> EventFactory<TwoPhaseEvent<T>> factory(EventFactory<T> outputFactory) {
//         return () -> new TwoPhaseEvent<>(outputFactory.newInstance());
//     }
// 
//     private final LineBytesBuffer input = new LineBytesBuffer();
//     private long lineNo;
//     private final T output;
// 
//     public TwoPhaseEvent(T output) {
//         this.output = output;
//     }
// 
//     public LineBytesBuffer input() {
//         return input;
//     }
// 
//     public T output() {
//         return output;
//     }
// 
//     public TwoPhaseEvent<T> lineNo(long lineNo) {
//         this.lineNo = lineNo;
//         return this;
//     }
// 
//     public long lineNo() {
//         return lineNo;
//     }
// }


import com.google.inject.Inject;
import com.google.inject.name.Named;
import com.lmax.disruptor.dsl.Disruptor;
import io.scaledml.core.BaseDisruptorRunner;
import io.scaledml.core.outputformats.OutputFormat;
import io.scaledml.core.TwoPhaseEvent;
import java.io.IOException;
import java.nio.file.Path;
import java.util.Optional;
package io.scaledml.ftrl;




public class FtrlProximalRunner extends BaseDisruptorRunner {
    private FtrlProximalModel model;
    private Path outputForModelPath;
    private OutputFormat outputFormat;

    protected void afterDisruptorProcessed() throws IOException {
        outputFormat.close();
        if (outputForModelPath != null) {
            FtrlProximalModel.saveModel(model, outputForModelPath);
        }
    }

    @Inject
// Path: fast-ftrl-proximal/src/main/java/io/scaledml/core/BaseDisruptorRunner.java
// public abstract class BaseDisruptorRunner {
//     private Disruptor<? extends TwoPhaseEvent<?>> disruptor;
//     private Supplier<InputStream> inputStreamFactory;
//     private boolean skipFirst;
//     private Phaser phaser;
// 
//     public void process() throws IOException {
//         try (FastBufferedInputStream stream = new FastBufferedInputStream(inputStreamFactory.get())) {
//             Preconditions.checkArgument(phaser.getRegisteredParties() == 0);
//             phaser.register();
//             disruptor.start();
//             RingBuffer<? extends TwoPhaseEvent> ringBuffer = disruptor.getRingBuffer();
//             long cursor = ringBuffer.next();
//             LineBytesBuffer buffer = ringBuffer.get(cursor).input();
//             long lineNo = 0;
//             ringBuffer.get(cursor).lineNo(lineNo);
//             boolean needToSkipNext = skipFirst;
//             while (buffer.readLineFrom(stream)) {
//                 if (needToSkipNext) {
//                     needToSkipNext = false;
//                     continue;
//                 }
//                 lineNo++;
//                 ringBuffer.publish(cursor);
//                 cursor = ringBuffer.next();
//                 buffer = ringBuffer.get(cursor).input();
//                 ringBuffer.get(cursor).lineNo(lineNo);
//             }
//             disruptor.shutdown();
//             phaser.arriveAndAwaitAdvance();
//             phaser.arriveAndDeregister();
//         } finally {
//             afterDisruptorProcessed();
//         }
//     }
// 
//     protected abstract void afterDisruptorProcessed() throws IOException;
// 
//     protected void setDisruptor(Disruptor<? extends TwoPhaseEvent<?>> disruptor) {
//         this.disruptor = disruptor;
//     }
// 
//     @Inject
//     public void setInputStreamFactory(Supplier<InputStream> inputStreamFactory) {
//         this.inputStreamFactory = inputStreamFactory;
//     }
// 
//     @Inject
//     public void setSkipFirst(@Named("skipFirst") boolean skipFirst) {
//         this.skipFirst = skipFirst;
//     }
// 
//     @Inject
//     public void setPhaser(Phaser phaser) {
//         this.phaser = phaser;
//     }
// }
// 
// Path: fast-ftrl-proximal/src/main/java/io/scaledml/core/outputformats/OutputFormat.java
// public interface OutputFormat extends Closeable {
//     void emit(SparseItem item, double prediction);
// }
// 
// Path: fast-ftrl-proximal/src/main/java/io/scaledml/core/TwoPhaseEvent.java
// public class TwoPhaseEvent<T> {
//     public static <T> EventFactory<TwoPhaseEvent<T>> factory(EventFactory<T> outputFactory) {
//         return () -> new TwoPhaseEvent<>(outputFactory.newInstance());
//     }
// 
//     private final LineBytesBuffer input = new LineBytesBuffer();
//     private long lineNo;
//     private final T output;
// 
//     public TwoPhaseEvent(T output) {
//         this.output = output;
//     }
// 
//     public LineBytesBuffer input() {
//         return input;
//     }
// 
//     public T output() {
//         return output;
//     }
// 
//     public TwoPhaseEvent<T> lineNo(long lineNo) {
//         this.lineNo = lineNo;
//         return this;
//     }
// 
//     public long lineNo() {
//         return lineNo;
//     }
// }


// Path: fast-ftrl-proximal/src/main/java/io/scaledml/ftrl/FtrlProximalRunner.java
import com.google.inject.Inject;
import com.google.inject.name.Named;
import com.lmax.disruptor.dsl.Disruptor;
import io.scaledml.core.BaseDisruptorRunner;
import io.scaledml.core.outputformats.OutputFormat;
import io.scaledml.core.TwoPhaseEvent;
import java.io.IOException;
import java.nio.file.Path;
import java.util.Optional;

package io.scaledml.ftrl;




public class FtrlProximalRunner extends BaseDisruptorRunner {
    private FtrlProximalModel model;
    private Path outputForModelPath;
    private OutputFormat outputFormat;

    protected void afterDisruptorProcessed() throws IOException {
        outputFormat.close();
        if (outputForModelPath != null) {
            FtrlProximalModel.saveModel(model, outputForModelPath);
        }
    }

    @Inject
    public FtrlProximalRunner disruptor(@Named("disruptor") Disruptor<? extends TwoPhaseEvent<?>> disruptor) {