/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * license agreements; and to You under the Apache License, version 2.0:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * This file is part of the Apache Pekko project, which was derived from Akka.
 */

/*
 * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
 */

package org.apache.pekko.stream.connectors.file.impl;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import org.apache.pekko.annotation.InternalApi;
import org.apache.pekko.stream.Attributes;
import org.apache.pekko.stream.Outlet;
import org.apache.pekko.stream.SourceShape;
import org.apache.pekko.stream.stage.*;
import org.apache.pekko.util.ByteString;
import scala.concurrent.duration.FiniteDuration;
import scala.util.Failure;
import scala.util.Success;
import scala.util.Try;

/**
 * INTERNAL API
 *
 * <p>Read the entire contents of a file, and then when the end is reached, keep reading newly
 * appended data. Like the unix command `tail -f`.
 *
 * <p>Aborting the stage can be done by combining with a `org.apache.pekko.stream.KillSwitch`
 *
 * <p>To use the stage from Scala see the factory methods in
 * `org.apache.pekko.stream.connectors.file.scaladsl.FileTailSource`
 */
@InternalApi
public final class FileTailSource extends GraphStage<SourceShape<ByteString>> {

  private final Path path;
  private final int maxChunkSize;
  private final long startingPosition;
  private final FiniteDuration pollingInterval;
  private final Outlet<ByteString> out = Outlet.create("FileTailSource.out");
  private final SourceShape<ByteString> shape = SourceShape.of(out);

  // this is stateless, so can be shared among instances
  private static final CompletionHandler<Integer, AsyncCallback<Try<Integer>>> completionHandler =
      new CompletionHandler<Integer, AsyncCallback<Try<Integer>>>() {
        @Override
        public void completed(Integer result, AsyncCallback<Try<Integer>> attachment) {
          attachment.invoke(new Success<>(result));
        }

        @Override
        public void failed(Throwable exc, AsyncCallback<Try<Integer>> attachment) {
          attachment.invoke(new Failure<>(exc));
        }
      };

  public FileTailSource(
      Path path, int maxChunkSize, long startingPosition, FiniteDuration pollingInterval) {
    this.path = path;
    this.maxChunkSize = maxChunkSize;
    this.startingPosition = startingPosition;
    this.pollingInterval = pollingInterval;
  }

  @Override
  public SourceShape<ByteString> shape() {
    return shape;
  }

  @Override
  public GraphStageLogic createLogic(Attributes inheritedAttributes) throws IOException {
    if (!Files.exists(path))
      throw new IllegalArgumentException("Path '" + path + "' does not exist");
    if (Files.isDirectory(path))
      throw new IllegalArgumentException("Path '" + path + "' cannot be tailed, it is a directory");
    if (!Files.isReadable(path))
      throw new IllegalArgumentException("No read permission for '" + path + "'");

    return new TimerGraphStageLogic(shape) {
      private final ByteBuffer buffer = ByteBuffer.allocate(maxChunkSize);
      private final AsynchronousFileChannel channel =
          AsynchronousFileChannel.open(path, StandardOpenOption.READ);

      private long position = startingPosition;
      private AsyncCallback<Try<Integer>> chunkCallback;

      {
        setHandler(
            out,
            new AbstractOutHandler() {
              @Override
              public void onPull() throws Exception {
                doPull();
              }
            });
      }

      @Override
      public void preStart() {
        chunkCallback =
            createAsyncCallback(
                (tryInteger) -> {
                  if (tryInteger.isSuccess()) {
                    int readBytes = tryInteger.get();
                    if (readBytes > 0) {
                      buffer.flip();
                      push(out, ByteString.fromByteBuffer(buffer));
                      position += readBytes;
                      buffer.clear();
                    } else {
                      // hit end, try again in a while
                      scheduleOnce("poll", pollingInterval);
                    }

                  } else {
                    failStage(tryInteger.failed().get());
                  }
                });
      }

      @Override
      public void onTimer(Object timerKey) {
        doPull();
      }

      private void doPull() {
        channel.read(buffer, position, chunkCallback, completionHandler);
      }

      @Override
      public void postStop() {
        try {
          if (channel.isOpen()) channel.close();
        } catch (Exception ex) {
          // Remove when #21168 is fixed
          throw new RuntimeException(ex);
        }
      }
    };
  }
}
