bigtop-master/bigtop-bigpetstore/bigpetstore-transaction-queue/src/main/java/org/apache/bigtop/bigpetstore/qstream/FileLoadGen.java
package org.apache.bigtop.bigpetstore.qstream;

import com.github.rnowling.bps.datagenerator.datamodels.Transaction;

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Stack;
import java.util.concurrent.LinkedBlockingQueue;

/*
*  Licensed to the Apache Software Foundation (ASF) under one or more
*  contributor license agreements.  See the NOTICE file distributed with
*  this work for additional information regarding copyright ownership.
*  The ASF licenses this file to You 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.
*/
public class FileLoadGen extends LoadGen{

    Path path;

    public FileLoadGen(int nStores, int nCustomers, double simulationLength, long seed, Path outputDir) throws Throwable {
        super(nStores, nCustomers, simulationLength, seed);
        path=outputDir;
    }

    public LinkedBlockingQueue<Transaction> startWriteQueue(final int milliseconds){
        if(! path.toFile().isDirectory()) {
            throw new RuntimeException("Input for the queue Should be a directory! Files will be transactions0.txt, transactions1.txt, and so on.");
        }

        /**
         * Write queue.   Every 5 seconds, write
         */
        final LinkedBlockingQueue<Transaction> transactionQueue = new LinkedBlockingQueue<Transaction>(getQueueSize());
        new Thread(){
            @Override
            public void run() {
                int fileNumber=0;
                while(true){
                    waitFor(milliseconds, transactionQueue);
                    System.out.println("Clearing " + transactionQueue.size() + " elements");
                    Stack<Transaction> transactionsToWrite = new Stack<Transaction>();
                    transactionQueue.drainTo(transactionsToWrite);
                    StringBuffer lines = new StringBuffer();
                    try{
                        while(!transactionsToWrite.isEmpty()){
                            lines.append(Utils.toJson(transactionsToWrite.pop())+"\n");
                            total++;
                        }
                        Path outputFile = Paths.get(path.toFile().getAbsolutePath(), "/transactions" + fileNumber++ + ".txt");
                        Files.write(outputFile, lines.toString().getBytes());
                        System.out.println("WRITING FILE to " + outputFile.toFile().length() + "bytes -> " + outputFile.toFile().getAbsolutePath());
                    }
                    catch(Throwable t){
                        t.printStackTrace();
                    }
                    System.out.println(
                            "TRANSACTIONS SO FAR " + total++ +" RATE " + (total/((System.currentTimeMillis()-startTime)/1000) + " per second "));
                }
            }
        }.start();
        return transactionQueue;
    }

}
