package com.chrisalbright;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.IntBuffer;
import java.nio.channels.FileChannel;
import java.util.Optional;

public class QueueFile<T> implements AutoCloseable {

  public static final int HEADER_SIZE = Integer.BYTES;
  private final RandomAccessFile raf;
  private final Converter<T> converter;
  private final int maxFileSize;
  private final FileChannel channel;
  private int readPosition = 0;
  private final IntBuffer readPositionBuffer;

  public QueueFile(File raf, Converter<T> converter) throws FileNotFoundException {
    this(raf, converter, 100 * 1024);
  }

  public QueueFile(File raf, Converter<T> converter, int maxFileSize) throws FileNotFoundException {
    this.raf = new RandomAccessFile(raf, "rw");
    this.converter = converter;
    this.maxFileSize = maxFileSize;
    this.channel = this.raf.getChannel();
    try {
      readPositionBuffer = channel.map(FileChannel.MapMode.READ_WRITE, 0, Integer.BYTES).asIntBuffer();
      readPosition = getReadPosition();
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  public void commit() {
    synchronized (readPositionBuffer) {
      readPositionBuffer.put(readPosition);
      readPositionBuffer.flip();
    }
  }

  private int getReadPosition() {
    int i = readPositionBuffer.get();
    if (i < HEADER_SIZE) {
      i = HEADER_SIZE;
    }
    readPositionBuffer.flip();
    return i;
  }


  @Override
  public void close() throws IOException {
    this.channel.close();
    this.raf.close();
  }

  public Optional<T> fetch() throws IOException {
    long length = 0;
    synchronized (channel) {
      length = channel.size() - readPosition;
    }

    ByteBuffer buffer;
    synchronized (this) {
      buffer = channel.map(FileChannel.MapMode.READ_ONLY, readPosition, length);
    }

    if (buffer.limit() <= 0) {
      return Optional.empty();
    }

    int byteLength = buffer.getInt();
    byte[] data = new byte[byteLength];
    buffer.get(data);

    synchronized (this) {
      readPosition += buffer.position();
    }

    return Optional.of(converter.fromBytes(data));
  }

  public boolean push(T val) throws IOException {
    byte[] bytes = converter.toBytes(val);
    ByteBuffer buffer;
    synchronized (channel) {
      int dataSize = bytes.length + Integer.BYTES;
      long channelSize = channel.size();
      if ((channelSize + dataSize) > maxFileSize) {
        return false;
      }
      buffer = channel.map(FileChannel.MapMode.READ_WRITE, channelSize, dataSize);
    }
    buffer.putInt(bytes.length);
    buffer.put(bytes);
    return true;
  }

  public static interface Converter<T> {
    public byte[] toBytes(T val);

    public T fromBytes(byte[] data);
  }
}

@@ -1245,50 +1245,8 @@
 ) %7B%0A
-    synchronized (readPositionBuffer) %7B%0A  
     
@@ -1283,18 +1283,16 @@
 ition);%0A
-  
     read
@@ -1314,22 +1314,16 @@
 flip();%0A
-    %7D%0A
   %7D%0A%0A  p
@@ -1681,39 +1681,8 @@
  0;%0A
-    synchronized (channel) %7B%0A  
     
@@ -1720,22 +1720,16 @@
 osition;
-%0A    %7D
 %0A%0A    By
@@ -1749,36 +1749,8 @@
 er;%0A
-    synchronized (this) %7B%0A  
     
@@ -1823,22 +1823,16 @@
 length);
-%0A    %7D
 %0A%0A    if
@@ -1999,36 +1999,8 @@
 );%0A%0A
-    synchronized (this) %7B%0A  
     
@@ -2033,22 +2033,16 @@
 ition();
-%0A    %7D
 %0A%0A    re
@@ -2211,39 +2211,8 @@
 er;%0A
-    synchronized (channel) %7B%0A  
     
@@ -2256,18 +2256,16 @@
 .BYTES;%0A
-  
     long
@@ -2299,18 +2299,16 @@
 ();%0A    
-  
 if ((cha
@@ -2347,26 +2347,24 @@
 ze) %7B%0A      
- 
- 
 return false
@@ -2369,22 +2369,18 @@
 se;%0A    
-  %7D%0A  
+%7D%0A
     buff
@@ -2452,22 +2452,16 @@
 aSize);%0A
-    %7D%0A
     buff

remove synchronized blocks until ready to make threadsafe