export JAVA_HOME=<Path of Java install Dir>

package chapter.five.kinesis.admin;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.services.kinesis.AmazonKinesisClient;
import com.amazonaws.services.kinesis.model.*;

public class ManageKinesisStreams {

  private AmazonKinesisClient kClient;

  /**
   * Constructor which initialize the Kinesis Client for working with the 
   * Kinesis streams.
   */
  public ManageKinesisStreams() {

    //Initialize the AWSCredentials taken from the profile configured in the Eclipse
    //replace "kinesisCred" with the profile configured in your Eclipse or leave blank to sue default.
    //Ensure that Access and Secret key are present in credentials file
    //Default location of credentials file is $USER_HOME/.aws/credentials
    AWSCredentials cred = new ProfileCredentialsProvider("kinesisCred")
        .getCredentials();
    System.out.println("Access Key = "+ cred.getAWSAccessKeyId());
    System.out.println("Secret Key = " + cred.getAWSSecretKey());
    kClient = new AmazonKinesisClient(cred);

  }

  /**
   * Create a Kinesis Stream with the given name and the shards.
   * @param streamName - Name of the Stream
   * @param shards - Number of Shards for the given Stream
   */
  public void createKinesisStream(String streamName, int shards) {

    System.out.println("Creating new Stream = '"+streamName+"', with Shards = "+shards);
    //Check and create stream only if does not exist.
    if (!checkStreamExists(streamName)) {
      //CreateStreamRequest for creating Kinesis Stream
      CreateStreamRequest createStreamRequest = new CreateStreamRequest();
      createStreamRequest.setStreamName(streamName);
      createStreamRequest.setShardCount(shards);
      kClient.createStream(createStreamRequest);

      try {
        //Sleep for 30 seconds so that stream is initialized and created
        Thread.sleep(30000);

      } catch (InterruptedException e) {
        //No need to Print Anything
      }
    }
  }

  /**
   * Checks and delete a given Kinesis Stream
   *    @param streamName
   *            - Name of the Stream
   */
  public void deleteKinesisStream(String streamName) {

    //Check and delete stream only if exists.
    if (checkStreamExists(streamName)) {
      kClient.deleteStream(streamName);
      System.out.println("Deleted the Kinesis Stream = '" + streamName+"'");
      return;
    }

    System.out.println("Stream does not exists = " + streamName);

  }

  /**
   * Utility Method which checks whether a given Kinesis Stream Exists or Not
   * @param streamName - Name of the Stream
   * @return - True in case Stream already exists else False
   */
  public boolean checkStreamExists(String streamName) {

    try {
      //DescribeStreamRequest for Describing and checking the 
      //existence of given Kinesis Streams.
      DescribeStreamRequest desc = new DescribeStreamRequest();
      desc.setStreamName(streamName);
      DescribeStreamResult result = kClient.describeStream(desc);

      System.out.println("Kinesis Stream '" +streamName+ "' already exists...");
      System.out.println("Status of '"+ streamName + "' = "
          + result.getStreamDescription().getStreamStatus());

    } catch (ResourceNotFoundException exception) {
      System.out.println("Stream '"+streamName+"' does Not exists...Need to create One");
      return false;
    }

    return true;  }}

package chapter.five.kinesis.producers;

import java.io.*;
import java.nio.ByteBuffer;
import java.util.ArrayList;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.services.kinesis.AmazonKinesisClient;
import com.amazonaws.services.kinesis.model.*;

public class AWSChicagoCrimesProducers{

private AmazonKinesisClient kClient;
  
//Location of the file from where we need to read the data
private String filePath="ChicagoCrimes-Aug-2015.csv";

/**
 * Constructor which initialize the Kinesis Client for working with the
 * Kinesis streams.
   */
public AWSChicagoCrimesProducers() {
// Initialize the AWSCredentials taken from the profile configured in
// the Eclipse replace "kinesisCred" with the profile //configured in your Eclipse or leave blank to use default.
// Ensure that Access and Secret key are present in credentials file
// Default location of credentials file is $USER_HOME/.aws/credentials
AWSCredentials cred = new ProfileCredentialsProvider("kinesisCred")
        .getCredentials();
  kClient = new AmazonKinesisClient(cred);
  }

  /**
   * Read Each record of the input file and Submit each record to Amazon Kinesis Streams.  
   * @param streamName - Name of the Stream.
   */
  public void readSingleRecordAndSubmit(String streamName) {

    String data = "";
try (BufferedReader br = new BufferedReader(
      new FileReader(new File(filePath)))) {
      //Skipping first line as it has headers;
      br.readLine();
      //Read Complete file - Line by Line
      while ((data = br.readLine()) != null) {
        //Create Record Request
        PutRecordRequest putRecordRequest = new PutRecordRequest();
        putRecordRequest.setStreamName(streamName);
        putRecordRequest.setData(ByteBuffer.wrap((data.getBytes())));
        //Data would be partitioned by the IUCR Codes, which is 5 column in the record
        String IUCRcode = (data.split(","))[4]; 
        putRecordRequest.setPartitionKey(IUCRcode);
//Finally Submit the records with Kinesis Client Object
System.out.println("Submitting Record = "+data);
kClient.putRecord(putRecordRequest);
//Sleep for half a second before we read and submit next record.
    Thread.sleep(500);
  }
} catch (Exception e) {
  //Print exceptions, in case any
  e.printStackTrace();
    }

  }
  
  /**
   * Read Data line by line by Submit to Kinesis Streams in the Batches. 
   * @param streamName - Name of Stream
   * @param batchSize - Batch Size
   */
  public void readAndSubmitBatch(String streamName, int batchSize) {

    String data = "";
    try (BufferedReader br = new BufferedReader(
        new FileReader(new File(filePath)))) {

    //Skipping first line as it has headers;
    br.readLine();
  //Counter to keep track of size of Batch
  int counter = 0;
  //Collection which will contain the batch of records
      ArrayList<PutRecordsRequestEntry> recordRequestEntryList = new ArrayList<PutRecordsRequestEntry>();
      while ((data = br.readLine()) != null) {
//Read Data and Create Object of PutRecordsRequestEntry
PutRecordsRequestEntry entry = new PutRecordsRequestEntry();
        entry.setData(ByteBuffer.wrap((data.getBytes())));
        //Data would be partitioned by the IUCR Codes, which is 5 column in the record
        String IUCRcode = (data.split(","))[4];
        entry.setPartitionKey(IUCRcode);
        //Add the record the Collection
        recordRequestEntryList.add(entry);
        //Increment the Counter
        counter++;

        //Submit Records in case Batch size is reached.
        if (counter == batchSize) {
          PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
          putRecordsRequest.setRecords(recordRequestEntryList);
          putRecordsRequest.setStreamName(streamName);
          //Finally Submit the records with Kinesis Client Object
          System.out.println("Submitting Records = "+recordRequestEntryList.size());
          kClient.putRecords(putRecordsRequest);
          //Reset the collection and Counter/Batch
          recordRequestEntryList = new ArrayList<PutRecordsRequestEntry>();
          counter = 0;
          //Sleep for half a second before processing another batch
          Thread.sleep(500);
        }
      }
    } catch (Exception e) {
      //Print all exceptions
      e.printStackTrace();
    }
  }
}

package chapter.five.kinesis;

import chapter.five.kinesis.admin.ManageKinesisStreams;
import chapter.five.kinesis.consumers.*;

public class MainRunConsumers {

//This Stream will be used by the producers/consumers using AWS SDK
public static String streamNameAWS = "AWSStreamingService";
  
  public static void main(String[] args) {

    //Using AWS Native API's
    AWSChicagoCrimesConsumers consumers = new AWSChicagoCrimesConsumers();
    consumers.consumeAndAlert(streamNameAWS);
        
//Enable only when you want to Delete the streams
ManageKinesisStreams streams = new ManageKinesisStreams();
    //streams.deleteKinesisStream(streamNameAWS);
    //streams.deleteKinesisStream(streamName);
}}

package chapter.five.kinesis;

import chapter.five.kinesis.admin.ManageKinesisStreams;
import chapter.five.kinesis.producers.*;

public class MainRunProducers {

//This Stream will be used by the producers/consumers using AWS SDK
  public static String streamNameAWS = "AWSStreamingService";
  
public static void main(String[] args) {
ManageKinesisStreams streams = new ManageKinesisStreams();
streams.createKinesisStream(streamNameAWS, 1);
//Using AWS Native API's
AWSChicagoCrimesProducers producers = new AWSChicagoCrimesProducers();

//Read and Submit record by record
//producers.readSingleRecordAndSubmit(streamName);
//Submit the records in Batches of 10
producers.readAndSubmitBatch(streamNameAWS, 10);
  
  //Enable only when you want to Delete the streams
  //streams.deleteKinesisStream(streamNameAWS);
  }}