/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed 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 from
 * 
 *     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 edu.uci.ics.hyracks.control.nc.dataset;

import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.logging.Level;
import java.util.logging.Logger;

import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataset.IDatasetPartitionManager;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.api.io.IFileHandle;
import edu.uci.ics.hyracks.api.io.IIOManager;
import edu.uci.ics.hyracks.api.partitions.IPartition;
import edu.uci.ics.hyracks.comm.channels.NetworkOutputChannel;

public class DatasetPartitionWriter implements IFrameWriter, IPartition {
    private static final Logger LOGGER = Logger.getLogger(DatasetPartitionWriter.class.getName());

    private static final String FILE_PREFIX = "result_";

    private final IHyracksTaskContext ctx;

    private final IDatasetPartitionManager manager;

    private final int partition;

    private final Executor executor;

    private FileReference fRef;

    private IFileHandle handle;

    private long size;

    private boolean failed;

    public DatasetPartitionWriter(IHyracksTaskContext ctx, IDatasetPartitionManager manager, int partition,
            Executor executor) {
        this.ctx = ctx;
        this.manager = manager;
        this.partition = partition;
        this.executor = executor;
    }

    @Override
    public void open() throws HyracksDataException {
        if (LOGGER.isLoggable(Level.INFO)) {
            LOGGER.info("open(" + partition + ")");
        }
        fRef = manager.getFileFactory().createUnmanagedWorkspaceFile(FILE_PREFIX + String.valueOf(partition));
        handle = ctx.getIOManager().open(fRef, IIOManager.FileReadWriteMode.READ_WRITE,
                IIOManager.FileSyncMode.METADATA_ASYNC_DATA_ASYNC);
        size = 0;
        failed = false;
    }

    @Override
    public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
        size += ctx.getIOManager().syncWrite(handle, size, buffer);
    }

    @Override
    public void fail() throws HyracksDataException {
        failed = true;
    }

    @Override
    public void close() throws HyracksDataException {
        if (LOGGER.isLoggable(Level.INFO)) {
            LOGGER.info("close(" + partition + ")");
        }
        /* TODO(madhusudancs): Do something more intelligent here than closing the file handle because read still
         * wants it :-P
         */
        // ctx.getIOManager().close(handle);
    }

    @Override
    public IHyracksTaskContext getTaskContext() {
        return ctx;
    }

    private long read(long offset, ByteBuffer buffer) throws HyracksDataException {
        return ctx.getIOManager().syncRead(handle, offset, buffer);
    }

    @Override
    public void writeTo(final IFrameWriter writer) {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                NetworkOutputChannel channel = (NetworkOutputChannel) writer;
                channel.setTaskContext(ctx);
                try {
                    channel.open();
                    try {
                        long offset = 0;
                        ByteBuffer buffer = ctx.allocateFrame();
                        while (true) {
                            buffer.clear();
                            long size = read(offset, buffer);
                            if (size < 0) {
                                break;
                            } else if (size < buffer.capacity()) {
                                throw new HyracksDataException("Premature end of file");
                            }
                            offset += size;
                            buffer.flip();
                            channel.nextFrame(buffer);
                        }
                    } finally {
                        channel.close();
                    }
                } catch (HyracksDataException e) {
                    throw new RuntimeException(e);
                }
            }
        });
    }

    @Override
    public boolean isReusable() {
        return true;
    }

    @Override
    public void deallocate() {

    }
}

@@ -744,16 +744,66 @@
 ecutor;%0A
+import java.util.concurrent.atomic.AtomicBoolean;%0A
 import j
@@ -826,16 +826,16 @@
 .Level;%0A
-
 import j
@@ -1756,24 +1756,62 @@
  executor;%0A%0A
+    private final AtomicBoolean eos;%0A%0A
     private 
@@ -2184,16 +2184,56 @@
 ecutor;%0A
+        eos = new AtomicBoolean(false);%0A
     %7D%0A%0A 
@@ -2742,24 +2742,37 @@
 %0A    public 
+synchronized 
 void nextFra
@@ -2883,32 +2883,53 @@
  size, buffer);%0A
+        notifyAll();%0A
     %7D%0A%0A    @Over
@@ -3037,24 +3037,37 @@
 %0A    public 
+synchronized 
 void close()
@@ -3197,32 +3197,76 @@
 %22)%22);%0A        %7D%0A
+        eos.set(true);%0A        notifyAll();%0A
         /* TODO(
@@ -3550,24 +3550,37 @@
     private 
+synchronized 
 long read(lo
@@ -3631,32 +3631,243 @@
 DataException %7B%0A
+        while (offset %3E= size && !eos.get()) %7B%0A            try %7B%0A                wait();%0A            %7D catch (InterruptedException e) %7B%0A                throw new HyracksDataException(e);%0A            %7D%0A        %7D%0A
         return c

Synchronize reads and writes to not let the reader attempt to read the file before it is written by the writer.