package com.epickrram.workshop.perf.app.processors;

//////////////////////////////////////////////////////////////////////////////////
//   Copyright 2015   Mark Price     mark at epickrram.com                      //
//                                                                              //
//   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 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.                                             //
//////////////////////////////////////////////////////////////////////////////////


import com.epickrram.workshop.perf.app.message.Packet;
import com.epickrram.workshop.perf.config.CommandLineArgs;
import com.epickrram.workshop.perf.support.NanoTimer;
import com.lmax.disruptor.RingBuffer;
import org.performancehints.SpinHint;

import java.io.File;
import java.io.IOException;
import java.nio.MappedByteBuffer;
import java.util.Date;
import java.util.concurrent.TimeUnit;

import static com.epickrram.workshop.perf.support.Threads.THREADS;
import static java.nio.channels.FileChannel.MapMode.READ_ONLY;
import static java.nio.channels.FileChannel.open;
import static java.nio.file.StandardOpenOption.READ;

public final class InputReader
{
    private final RingBuffer<Packet> messageSink;
    private final NanoTimer nanoTimer;
    private final CommandLineArgs commandLineArgs;
    private final int numberOfIterations;

    public InputReader(final RingBuffer<Packet> messageSink, final NanoTimer nanoTimer, final CommandLineArgs commandLineArgs)
    {
        this.messageSink = messageSink;
        this.nanoTimer = nanoTimer;
        this.commandLineArgs = commandLineArgs;
        this.numberOfIterations = commandLineArgs.getNumberOfIterations();
    }

    public void processFiles()
    {
        System.out.println("Producer thread has pid: " + THREADS.getCurrentThreadId());
        for(int i = 0; i < numberOfIterations; i++)
        {
            if(i == commandLineArgs.getNumberOfWarmups())
            {
                System.out.println("Warm-up complete at " + new Date());
                System.out.println("Pausing for 10 seconds...");
                THREADS.sleep(10L, TimeUnit.SECONDS);
                System.out.println("Executing test at " + new Date());
            }
            processSingleFile(new File(commandLineArgs.getInputFile()), i == numberOfIterations - 1);
        }
    }

    private void processSingleFile(final File inputFile, final boolean isLastIteration)
    {
        try
        {
            final MappedByteBuffer mappedFile = open(inputFile.toPath(), READ).map(READ_ONLY, 0L, inputFile.length());
            mappedFile.load();

            for(int recordIndex = 0; recordIndex < commandLineArgs.getNumberOfRecords(); recordIndex++)
            {
                if(!messageSink.hasAvailableCapacity(1))
                {
                    throw new RuntimeException("RingBuffer full!");
                }
                final long sequence = messageSink.next();

                try
                {
                    final Packet packet = messageSink.get(sequence);
                    packet.reset();
                    packet.setSequenceInFile(recordIndex);
                    packet.setSequence(sequence);
                    final boolean isLastInFile = recordIndex == commandLineArgs.getNumberOfRecords() - 1;
                    packet.setLastInFile(isLastInFile);
                    packet.setReceivedNanoTime(nanoTimer.nanoTime());
                    packet.setLastInStream(isLastInFile && isLastIteration);
                    mappedFile.position(recordIndex * commandLineArgs.getRecordLength());
                    mappedFile.limit(mappedFile.position() + commandLineArgs.getRecordLength());
                    packet.getPayload().put(mappedFile);
                }
                finally
                {
                    messageSink.publish(sequence);
                }
                introduceMessagePublishDelay();
            }

        }
        catch (IOException e)
        {
            throw new RuntimeException("File operation failed", e);
        }
    }

    private void introduceMessagePublishDelay()
    {
        final long stopSpinningAt = System.nanoTime() + TimeUnit.MICROSECONDS.toNanos(10L);
        while(System.nanoTime() < stopSpinningAt)
        {
            SpinHint.spinLoopHint();
        }
    }
}
@@ -1617,24 +1617,62 @@
 ByteBuffer;%0A
+import java.nio.channels.FileChannel;%0A
 import java.
@@ -3297,41 +3297,41 @@
     
+try(
 final 
-MappedByteBuffer mappedFile
+FileChannel fileChannel
  = o
@@ -3359,16 +3359,95 @@
 ), READ)
+)%0A            %7B%0A                final MappedByteBuffer mappedFile = fileChannel
 .map(REA
@@ -3471,32 +3471,36 @@
 File.length());%0A
+    
             mapp
@@ -3523,27 +3523,32 @@
             
+    
 for
+ 
 (int recordI
@@ -3628,34 +3628,42 @@
 ++)%0A            
-%7B%0A
+    %7B%0A    
                 
@@ -3664,16 +3664,17 @@
       if
+ 
 (!messag
@@ -3712,34 +3712,42 @@
                 
+  
-%7B%0A
+  %7B%0A    
                 
@@ -3806,34 +3806,42 @@
                 
-%7D%0A
+    %7D%0A    
                 
@@ -3891,32 +3891,36 @@
                 
+ 
+   
 try%0A            
@@ -3915,34 +3915,42 @@
                 
-%7B%0A
+    %7B%0A    
                 
@@ -4018,24 +4018,28 @@
             
+    
 packet.reset
@@ -4054,32 +4054,36 @@
                 
+ 
+   
 packet.setSequen
@@ -4117,32 +4117,36 @@
                 
+    
 packet.setSequen
@@ -4151,32 +4151,36 @@
 ence(sequence);%0A
+    
                 
@@ -4269,16 +4269,20 @@
 () - 1;%0A
+    
         
@@ -4341,32 +4341,36 @@
                 
+    
 packet.setReceiv
@@ -4399,24 +4399,28 @@
 anoTime());%0A
+    
             
@@ -4496,32 +4496,36 @@
                 
+    
 mappedFile.posit
@@ -4570,32 +4570,36 @@
 ecordLength());%0A
+    
                 
@@ -4695,24 +4695,28 @@
             
+    
 packet.getPa
@@ -4748,34 +4748,42 @@
                 
-%7D%0A
+    %7D%0A    
                 
@@ -4798,34 +4798,42 @@
                 
+   
+ 
 %7B%0A
+    
                 
@@ -4875,34 +4875,42 @@
                 
-%7D%0A
+    %7D%0A    
                 
@@ -4949,26 +4949,43 @@
             
+  
+  %7D%0A            
 %7D
-%0A
 %0A        %7D%0A 

fix file handle leak