KinesisScalaSample.scala

import java.io.File
import java.util


import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
import com.amazonaws.services.dynamodbv2.document.spec._
import com.amazonaws.services.dynamodbv2.document.utils.{NameMap, ValueMap}
import com.amazonaws.services.dynamodbv2.document._
import com.amazonaws.services.dynamodbv2.model._
import com.fasterxml.jackson.core.JsonFactory
import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
import collection.JavaConversions._

class DynamoDBTrial {
  val client = new AmazonDynamoDBClient()
    .withEndpoint("http://localhost:8000").asInstanceOf[AmazonDynamoDBClient]

  val dynamoDB = new DynamoDB(client)
  val tableName = "Movies"

  def tableCreate: Unit = {
  
  try {
      println("Attempting to create table please wait...")
      val table = dynamoDB.createTable(tableName,
        util.Arrays.asList(
          new KeySchemaElement("year", KeyType.HASH),  //Partition key
          new KeySchemaElement("title", KeyType.RANGE)), //Sort key
        util.Arrays.asList(
          new AttributeDefinition("year", ScalarAttributeType.N),
          new AttributeDefinition("title", ScalarAttributeType.S)),
        new ProvisionedThroughput(10L, 10L))
      table.waitForActive()
      println("Success.  Table status: " + table.getDescription().getTableStatus())

    } catch {
      case ex:Exception => {
        println("Unable to create table: ")
        println(ex.getMessage)
      }
    }
  }

  def tableInsert {
    val table = dynamoDB.getTable(tableName)
    val parser = new JsonFactory()
      .createParser(new File("/Users/Shibin/dynamodb_local_latest/moviedata.json"))

    val rootNode = new ObjectMapper().readTree(parser).asInstanceOf[JsonNode]
    val iter = rootNode.iterator()

    while (iter.hasNext) {
      val currentNode = iter.next()//.asInstanceOf[ObjectNode]

      val year = currentNode.path("year").asInt()
      val title = currentNode.path("title").asText()

      try {
        table.putItem(new Item()
          .withPrimaryKey("year", year, "title", title)
          .withJSON("info", currentNode.path("info").toString()))
        println("PutItem succeeded: " + year + " " + title)

      } catch {
        case ex:Exception => {
          println("Unable to add movie: " + year + " " + title)
          println(ex.getMessage())
        }
      }
      parser.close()
    }
  }

  def createNewItem: Unit = {
    val table = dynamoDB.getTable("Movies")

    val year = 2015
    val title = "The Big New Movie"

    val infoMap = new util.HashMap[String, Object]()
    infoMap.put("plot",  "Nothing happens at all.")
    infoMap.put("rating",  0.asInstanceOf[Object])

    try {
      println("Adding a new item...")
      val outcome = table.putItem(new Item().withPrimaryKey("year", year, "title", title).withMap("info", infoMap))

      println("PutItem succeeded:\n" + outcome.getPutItemResult())

    } catch {
      case ex:Exception => {
        println("Unable to add item: " + year + " " + title)
        println(ex.getMessage())
      }
    }
  }

  def readItem {
    val table = dynamoDB.getTable("Movies")

    val year = 2015
    val title = "The Big New Movie"

    val spec = new GetItemSpec()
      .withPrimaryKey("year", year, "title", title)

    try {
      println("Attempting to read the item...")
      val outcome = table.getItem(spec)
      println("GetItem succeeded: " + outcome)

    } catch {
      case ex:Exception => {
        println("Unable to read item: " + year + " " + title)
        println(ex.getMessage())
      }
    }
  }

  def updateItem {
    val table = dynamoDB.getTable("Movies")

    val year = 2015
    val title = "The Big New Movie"

    val updateItemSpec = new UpdateItemSpec()
      .withPrimaryKey("year", year, "title", title)
      .withUpdateExpression("set info.rating = :r, info.plot=:p, info.actors=:a")
      .withValueMap(new ValueMap()
        .withNumber(":r", 5.5)
        .withString(":p", "Everything happens all at once.")
        .withList(":a", util.Arrays.asList("Larry","Moe","Curly")))
      .withReturnValues(ReturnValue.UPDATED_NEW)

    try {
      println("Updating the item...")
      val outcome = table.updateItem(updateItemSpec)
      println("UpdateItem succeeded:\n" + outcome.getItem().toJSONPretty())

    } catch {
      case ex:Exception => {
        println("Unable to update item: " + year + " " + title)
        println(ex.getMessage)
      }
    }
  }

  def updateAtomic {
    val table = dynamoDB.getTable("Movies")

    val year = 2015
    val title = "The Big New Movie"

    val updateItemSpec = new UpdateItemSpec()
      .withPrimaryKey("year", year, "title", title)
      .withUpdateExpression("set info.rating = info.rating + :val")
      .withValueMap(new ValueMap()
        .withNumber(":val", 1))
      .withReturnValues(ReturnValue.UPDATED_NEW)

    try {
      println("Incrementing an atomic counter...")
      val outcome = table.updateItem(updateItemSpec)
      println("UpdateItem succeeded:\n" + outcome.getItem().toJSONPretty())

    } catch  {
      case ex:Exception =>
        println("Unable to update item: " + year + " " + title)
        println(ex.getMessage())
    }

  }

  def updateOnCondition {
    val table = dynamoDB.getTable("Movies")
    val year = 2015
    val title = "The Big New Movie"

    val updateItemSpec = new UpdateItemSpec()
      .withPrimaryKey(new PrimaryKey("year", year, "title", title))
      .withUpdateExpression("remove info.actors[0]")
      .withConditionExpression("size(info.actors) > :num")
      .withValueMap(new ValueMap().withNumber(":num", 2))
      .withReturnValues(ReturnValue.UPDATED_NEW)

    // Conditional update (we expect this to fail)
    try {
      println("Attempting a conditional update...")
      val outcome = table.updateItem(updateItemSpec)
      println("UpdateItem succeeded:\n" + outcome.getItem().toJSONPretty())

    } catch {
      case ex:Exception =>
        println("Unable to update item: " + year + " " + title)
        println(ex.getMessage())
    }
  }

  def deleteItem: Unit = {
    val table = dynamoDB.getTable("Movies")
    val year = 2015
    val title = "The Big New Movie"

    val deleteItemSpec = new DeleteItemSpec()
      .withPrimaryKey(new PrimaryKey("year", year, "title", title))
      .withConditionExpression("info.rating <= :val")
      .withValueMap(new ValueMap()
        .withNumber(":val", 7.0))

    // Conditional update (we expect this to fail)
    try {
      println("Attempting a conditional delete...")
      table.deleteItem(deleteItemSpec)
      println("DeleteItem succeeded")
    } catch {
      case ex:Exception =>
        println("Unable to delete item: " + year + " " + title)
        println(ex.getMessage())
    }
  }

  def queryItems: Unit = {
    val table = dynamoDB.getTable("Movies")

    val nameMap = new util.HashMap[String, String]()
    nameMap.put("#yr", "year")

    val valueMap = new util.HashMap[String, Object]()
    valueMap.put(":yyyy", 1985.asInstanceOf[Object])

    val querySpec = new QuerySpec()
      .withKeyConditionExpression("#yr = :yyyy")
      .withNameMap(nameMap)
      .withValueMap(valueMap)

    var items:ItemCollection[QueryOutcome] = null
    var item:Item = null
    var iterator:Iterator[Item]  = null
    try {
      println("Movies from 1985")
      items = table.query(querySpec)

      iterator = items.iterator()
      while (iterator.hasNext) {
        item  = iterator.next()
        println(item.getNumber("year") + ": "
          + item.getString("title"))
      }

    } catch { case ex:Exception =>
      println("Unable to query movies from 1985")
      println(ex.getMessage)
    }

    valueMap.put(":yyyy", 1992.asInstanceOf[Object])
    valueMap.put(":letter1", "A")
    valueMap.put(":letter2", "L")

    querySpec
      .withProjectionExpression(
        "#yr, title, info.genres, info.actors[0]")
      .withKeyConditionExpression(
        "#yr = :yyyy and title between :letter1 and :letter2")
      .withNameMap(nameMap).withValueMap(valueMap)

    try {
      System.out
        .println("Movies from 1992 - titles A-L, with genres and lead actor")
      items = table.query(querySpec)

      iterator = items.iterator()
      while (iterator.hasNext) {
        item = iterator.next()
        println(item.getNumber("year") + ": "
          + item.getString("title") + " " + item.getMap("info"))
      }

    } catch {case ex: Exception =>
      println("Unable to query movies from 1992:")
      println(ex.getMessage())
    }
  }

  def scanItems: Unit = {
    val table = dynamoDB.getTable("Movies")
    val scanSpec = new ScanSpec()
      .withProjectionExpression("#yr, title, info.rating")
      .withFilterExpression("#yr between :start_yr and :end_yr")
      .withNameMap(new NameMap().`with`("#yr",  "year"))
      .withValueMap(new ValueMap().withNumber(":start_yr", 1950).withNumber(":end_yr", 1959))

    try {
      val items = table.scan(scanSpec)

      val iter = items.iterator()
      while (iter.hasNext) {
        val item = iter.next()
        println(item.toString)
      }
    } catch {
      case ex:Exception =>
        println("Unable to scan the table:")
        println(ex.getMessage)
    }
  }
}