package au.gov.amsa.risky.format;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.zip.GZIPOutputStream;

import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;

import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.observables.GroupedObservable;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.Striped;

public final class BinaryFixesWriter {

	public static Observable<List<Fix>> writeFixes(final Func1<Fix, String> fileMapper,
	        Observable<Fix> fixes, int bufferSize, boolean zip) {
		return fixes
		// group by filename
		        .groupBy(fileMapper)
		        // buffer fixes by filename
		        .flatMap(buffer(bufferSize))
		        // write each list to a file
		        .doOnNext(writeFixList(fileMapper, zip));
	}

	private static Func1<GroupedObservable<String, Fix>, Observable<List<Fix>>> buffer(
	        final int bufferSize) {
		return new Func1<GroupedObservable<String, Fix>, Observable<List<Fix>>>() {
			@Override
			public Observable<List<Fix>> call(GroupedObservable<String, Fix> fileFixes) {
				return fileFixes.buffer(bufferSize);
			}
		};
	}

	private static Action1<List<Fix>> writeFixList(final Func1<Fix, String> fileMapper,
	        final boolean zip) {
		return new Action1<List<Fix>>() {

			@Override
			public void call(List<Fix> fixes) {
				if (fixes.size() == 0)
					return;
				String filename = fileMapper.call(fixes.get(0));
				writeFixes(fixes, new File(filename), true, zip);
			}

		};
	}

	private static final Striped<Lock> locks = Striped.lock(2000);

	public static void writeFixes(List<Fix> fixes, File file, boolean append, boolean zip) {
		Preconditions.checkArgument(!zip || !append, "cannot perform append and zip at same time");

		Lock lock = locks.get(file);

		OutputStream os = null;
		try {
			lock.lock();
			file.getParentFile().mkdirs();
			FileOutputStream fos = new FileOutputStream(file, append);
			OutputStream s;
			if (zip)
				s = new GZIPOutputStream(fos);
			else
				s = fos;
			os = new BufferedOutputStream(s);
			ByteBuffer bb = BinaryFixes.createFixByteBuffer();
			for (Fix fix : fixes) {
				bb.rewind();
				BinaryFixes.write(fix, bb);
				os.write(bb.array());
			}
		} catch (IOException e) {
			e.printStackTrace();
			throw new RuntimeException(e);
		} finally {
			try {
				if (os != null)
					try {
						os.close();
					} catch (IOException e) {
						// we care because we are writing
						throw new RuntimeException(e);
					}
			} finally {
				lock.unlock();
			}
		}
	}

	public static class ByMonth implements Func1<Fix, String> {

		private final String base;

		public ByMonth(File directory) {
			base = directory.getAbsolutePath();
		}

		@Override
		public String call(Fix fix) {
			DateTime d = new DateTime(fix.getTime(), DateTimeZone.UTC);
			int month = d.getMonthOfYear();
			int year = d.getYear();
			StringBuilder s = new StringBuilder();
			s.append(base);
			s.append(File.separator);
			s.append(year);
			s.append(File.separator);
			s.append(month);
			s.append(File.separator);
			s.append(fix.getMmsi());
			s.append(".track");
			return s.toString();
		}

	}

	public static class ByYear implements Func1<Fix, String> {

		private final String base;

		public ByYear(File directory) {
			base = directory.getAbsolutePath();
		}

		@Override
		public String call(Fix fix) {
			DateTime d = new DateTime(fix.getTime(), DateTimeZone.UTC);
			int year = d.getYear();
			StringBuilder s = new StringBuilder();
			s.append(base);
			s.append(File.separator);
			s.append(year);
			s.append(File.separator);
			s.append(fix.getMmsi());
			s.append(".track");
			return s.toString();
		}

	}

}

@@ -1722,32 +1722,291 @@
 %0A%09%09%09%7D%0A%0A%09%09%7D;%0A%09%7D%0A%0A
+%09private static final int NUMBER_FILE_LOCKS = 200;%0A%0A%09/**%0A%09 * A guava cache of locks so that file writes can happen simultaneously%0A%09 * (SSD) but not go too crazy (one lock per file). This maps files to a%0A%09 * limited set of locks using the file hash keys.%0A%09 */%0A
 %09private static 
@@ -2025,17 +2025,21 @@
 d%3CLock%3E 
-l
+fileL
 ocks = S
@@ -2054,12 +2054,25 @@
 ock(
-2000
+NUMBER_FILE_LOCKS
 );%0A%0A
@@ -2258,16 +2258,53 @@
 e%22);%0A%0A%09%09
+// get the lock for the file%0A%09%09final 
 Lock loc
@@ -2307,17 +2307,21 @@
  lock = 
-l
+fileL
 ocks.get
@@ -2370,20 +2370,82 @@
 %0A%09%09%09
-lock.lock();
+// open the lock for the file%0A%09%09%09lock.lock();%0A%0A%09%09%09// open an output stream
 %0A%09%09%09
@@ -2660,16 +2660,60 @@
 ream(s);
+%0A%0A%09%09%09// write the fixes to the output stream
 %0A%09%09%09Byte
@@ -3150,24 +3150,66 @@
 %7D finally %7B%0A
+%09%09%09%09// must unlock no matter what happens%0A
 %09%09%09%09lock.unl

use Striped locks