content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.distributedlog.fs;
  
  import static com.google.common.base.Preconditions.checkNotNull;
  
  import java.io.IOException;
  import java.io.InputStream;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.distributedlog.LogRecordWithDLSN;
  import org.apache.distributedlog.api.DistributedLogManager;
  import org.apache.distributedlog.api.LogReader;
  import org.apache.hadoop.fs.FSInputStream;
  
  /**
   * The input stream for a distributedlog stream.
   */
  @Slf4j
  class DLInputStream extends FSInputStream {
  
      private static final long REOPEN_READER_SKIP_BYTES = 4 * 1024 * 1024; // 4MB
  
      private static class RecordStream {
  
          private final InputStream payloadStream;
          private final LogRecordWithDLSN record;
  
          RecordStream(LogRecordWithDLSN record) {
              checkNotNull(record);
  
              this.record = record;
              this.payloadStream = record.getPayLoadInputStream();
          }
  
      }
  
      private static RecordStream nextRecordStream(LogReader reader) throws IOException {
          LogRecordWithDLSN record = reader.readNext(false);
          if (null != record) {
              return new RecordStream(record);
          }
          return null;
      }
  
      private final DistributedLogManager dlm;
      private LogReader reader;
      private long pos;
      private long lastPos;
      private RecordStream currentRecord = null;
  
      DLInputStream(DistributedLogManager dlm,
                    LogReader reader,
                    long startPos)
              throws IOException {
          this.dlm = dlm;
          this.reader = reader;
          this.pos = startPos;
          this.lastPos = readEndPos();
          seek(startPos);
      }
  
      @Override
      public void close() throws IOException {
          reader.close();
          dlm.close();
      }
  
      private long readEndPos() throws IOException {
          return dlm.getLastTxId();
      }
  
      //
      // FSInputStream
      //
  
      @Override
      public void seek(long pos) throws IOException {
          if (this.pos == pos) {
              return;
          }
  
          if (this.pos > pos || (pos - this.pos) >= REOPEN_READER_SKIP_BYTES) {
              // close the previous reader
              this.reader.close();
              this.reader = dlm.openLogReader(pos);
              this.currentRecord = null;
          }
  
          skipTo(pos);
      }
  
      private boolean skipTo(final long position) throws IOException {
          while (true) {
              if (null == currentRecord) {
                  currentRecord = nextRecordStream(reader);
              }
  
              if (null == currentRecord) { // the stream is empty now
                  return false;
              }
  
              long endPos = currentRecord.record.getTransactionId();
              if (endPos < position) {
                  currentRecord = nextRecordStream(reader);
                  this.pos = endPos;
                  continue;
              } else if (endPos == position){
                  // find the record, but we defer read next record when actual read happens
                  this.pos = position;
                  this.currentRecord = null;
                  return true;
              } else {
                  this.currentRecord.payloadStream.skip(
                      this.currentRecord.payloadStream.available() - (endPos - position));
                  this.pos = position;
                  return true;
              }
          }
      }
  
      @Override
      public long getPos() throws IOException {
          return this.pos;
      }
  
      @Override
      public boolean seekToNewSource(long targetPos) throws IOException {
          return false;
      }
  
      //
      // Input Stream
      //
  
      @Override
      public int read(byte[] b, final int off, final int len) throws IOException {
          int remaining = len;
          int numBytesRead = 0;
          while (remaining > 0) {
              if (null == currentRecord) {
                  currentRecord = nextRecordStream(reader);
              }
  
              if (null == currentRecord) {
                  if (numBytesRead == 0) {
                      return -1;
                  }
                  break;
              }
  
              int bytesLeft = currentRecord.payloadStream.available();
              if (bytesLeft <= 0) {
                  currentRecord.payloadStream.close();
                  currentRecord = null;
                  continue;
              }
  
              int numBytesToRead = Math.min(bytesLeft, remaining);
              int numBytes = currentRecord.payloadStream.read(b, off + numBytesRead, numBytesToRead);
              if (numBytes < 0) {
                  continue;
              }
              numBytesRead += numBytes;
              remaining -= numBytes;
          }
          return numBytesRead;
      }
  
      @Override
      public long skip(final long n) throws IOException {
          if (n <= 0L) {
              return 0L;
          }
  
          long remaining = n;
          while (true) {
              if (null == currentRecord) {
                  currentRecord = nextRecordStream(reader);
              }
  
              if (null == currentRecord) { // end of stream
                  return n - remaining;
              }
  
              int bytesLeft = currentRecord.payloadStream.available();
              long endPos = currentRecord.record.getTransactionId();
              if (remaining > bytesLeft) {
                  // skip the whole record
                  remaining -= bytesLeft;
                  this.pos = endPos;
                  this.currentRecord = nextRecordStream(reader);
                  continue;
              } else if (remaining == bytesLeft) {
                  this.pos = endPos;
                  this.currentRecord = null;
                  return n;
              } else {
                  currentRecord.payloadStream.skip(remaining);
                  this.pos = endPos - currentRecord.payloadStream.available();
                  return n;
              }
          }
      }
  
      @Override
      public int available() throws IOException {
          if (lastPos - pos == 0L) {
              lastPos = readEndPos();
          }
          return (int) (lastPos - pos);
      }
  
      @Override
      public boolean markSupported() {
          return false;
      }
  
      @Override
      public int read() throws IOException {
          byte[] data = new byte[1];
          int numBytes = read(data);
          if (numBytes <= 0) {
              return -1;
          }
          return data[0];
      }
  
  }
  