/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.worker.page;

import alluxio.client.file.CacheContext;
import alluxio.client.file.cache.CacheManager;
import alluxio.client.file.cache.CacheUsage;
import alluxio.client.file.cache.PageId;
import alluxio.file.ReadTargetBuffer;
import alluxio.network.protocol.databuffer.DataFileChannel;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.function.Supplier;

/**
 * Implementation of cache manager that stores cached data in byte arrays in memory.
 */
class ByteArrayCacheManager implements CacheManager {
  private final Map<PageId, byte[]> mPages;

  /**
   * Metrics for test validation.
   */
  long mPagesServed = 0;
  long mPagesCached = 0;

  ByteArrayCacheManager() {
    mPages = new HashMap<>();
  }

  @Override
  public void commitFile(String fileId) {
    throw new UnsupportedOperationException("commitFile method is unsupported. ");
  }

  @Override
  public boolean put(PageId pageId, ByteBuffer page, CacheContext cacheContext) {
    byte[] data = new byte[page.remaining()];
    page.get(data);
    mPages.put(pageId, data);
    mPagesCached++;
    return true;
  }

  @Override
  public int get(PageId pageId, int pageOffset, int bytesToRead, ReadTargetBuffer target,
      CacheContext cacheContext) {
    if (!mPages.containsKey(pageId)) {
      return 0;
    }
    mPagesServed++;
    target.writeBytes(mPages.get(pageId), pageOffset, bytesToRead);
    return bytesToRead;
  }

  @Override
  public int getAndLoad(PageId pageId, int pageOffset, int bytesToRead,
      ReadTargetBuffer buffer, CacheContext cacheContext,
      Supplier<byte[]> externalDataSupplier) {
    int bytesRead = get(pageId, pageOffset,
        bytesToRead, buffer, cacheContext);
    if (bytesRead > 0) {
      return bytesRead;
    }
    byte[] page = externalDataSupplier.get();
    if (page.length == 0) {
      return 0;
    }
    buffer.writeBytes(page, pageOffset, bytesToRead);
    put(pageId, page, cacheContext);
    return bytesToRead;
  }

  @Override
  public boolean delete(PageId pageId) {
    return mPages.remove(pageId) != null;
  }

  @Override
  public State state() {
    return State.READ_WRITE;
  }

  @Override
  public boolean append(PageId pageId, int appendAt, byte[] page, CacheContext cacheContext) {
    return false;
  }

  @Override
  public void deleteFile(String fileId) {
    mPages.keySet().removeIf(pageId -> pageId.getFileId().equals(fileId));
  }

  @Override
  public void deleteTempFile(String fileId) {
    mPages.keySet().removeIf(pageId -> pageId.getFileId().equals(fileId));
  }

  @Override
  public Optional<CacheUsage> getUsage() {
    return Optional.of(new Usage());
  }

  @Override
  public Optional<DataFileChannel> getDataFileChannel(PageId pageId, int pageOffset,
      int bytesToRead, CacheContext cacheContext) {
    return Optional.empty();
  }

  class Usage implements CacheUsage {
    @Override
    public Optional<CacheUsage> partitionedBy(PartitionDescriptor<?> partition) {
      return Optional.empty();
    }

    @Override
    public long used() {
      return mPages.values().stream().mapToInt(page -> page.length).sum();
    }

    @Override
    public long available() {
      return Integer.MAX_VALUE;
    }

    @Override
    public long capacity() {
      return Integer.MAX_VALUE;
    }
  }

  @Override
  public void close() throws Exception {
    // no-op
  }
}
