/**
 * Put item into FragmentCheckpoint table for the given input parameters
 *
 * @param streamName KVS Stream name
 * @param fragmentNumber Last processed fragment's fragment number
 * @param producerTime Last processed fragment's producer time
 * @param serverTime Last processed fragment's server time
 * @param updatedTime Time when the entry is going to be updated.
 */
public void putItem(final String streamName, final String fragmentNumber,
                     final Long producerTime, final Long serverTime, final Long updatedTime) {
    try {
        final Map<String,AttributeValue> item = new HashMap<>();
        item.put(KVS_STREAM_NAME, new AttributeValue().withS(streamName));
        item.put(FRAGMENT_NUMBER, new AttributeValue().withS(fragmentNumber));
        item.put(UPDATED_TIME, new AttributeValue().withN(updatedTime.toString()));
        item.put(PRODUCER_TIME, new AttributeValue().withN(producerTime.toString()));
        item.put(SERVER_TIME, new AttributeValue().withN(serverTime.toString()));
        final PutItemRequest putItemRequest = new PutItemRequest()
        {{
            setTableName(TABLE_NAME);
            setItem(item);
        }};

        final PutItemResult result = ddbClient.putItem(putItemRequest);
        log.info("Item saved : ", result.getAttributes());
    } catch (final Exception e) {
        log.warn("Error while putting item into the table!", e);
    }
}
 

public PutItemResult putItem(final PutItemRequest request) throws BackendException {
    setUserAgent(request);
    PutItemResult result;
    final int bytes = calculateItemSizeInBytes(request.getItem());
    getBytesHistogram(PUT_ITEM, request.getTableName()).update(bytes);
    final int wcu = computeWcu(bytes);
    timedWriteThrottle(PUT_ITEM, request.getTableName(), wcu);

    final Timer.Context apiTimerContext = getTimerContext(PUT_ITEM, request.getTableName());
    try {
        result = client.putItem(request);
    } catch (Exception e) {
        throw processDynamoDbApiException(e, PUT_ITEM, request.getTableName());
    } finally {
        apiTimerContext.stop();
    }
    meterConsumedCapacity(PUT_ITEM, result.getConsumedCapacity());

    return result;
}
 

public PutPointResult putPoint(PutPointRequest putPointRequest) {
	long geohash = S2Manager.generateGeohash(putPointRequest.getGeoPoint());
	long hashKey = S2Manager.generateHashKey(geohash, config.getHashKeyLength());
	String geoJson = GeoJsonMapper.stringFromGeoObject(putPointRequest.getGeoPoint());

	PutItemRequest putItemRequest = putPointRequest.getPutItemRequest();
	putItemRequest.setTableName(config.getTableName());

	AttributeValue hashKeyValue = new AttributeValue().withN(String.valueOf(hashKey));
	putItemRequest.getItem().put(config.getHashKeyAttributeName(), hashKeyValue);
	putItemRequest.getItem().put(config.getRangeKeyAttributeName(), putPointRequest.getRangeKeyValue());
	AttributeValue geohashValue = new AttributeValue().withN(Long.toString(geohash));
	putItemRequest.getItem().put(config.getGeohashAttributeName(), geohashValue);
	AttributeValue geoJsonValue = new AttributeValue().withS(geoJson);
	putItemRequest.getItem().put(config.getGeoJsonAttributeName(), geoJsonValue);

	PutItemResult putItemResult = config.getDynamoDBClient().putItem(putItemRequest);
	PutPointResult putPointResult = new PutPointResult(putItemResult);

	return putPointResult;
}
 

private static void putItem(Map<String, AttributeValue> item) {

        try {

            PutItemRequest putItemRequest = new PutItemRequest(TABLENAME, item);
            PutItemResult putItemResult = dynamoDBClient.putItem(putItemRequest);
            logger.info("Result: " + putItemResult);
        }
        catch (Exception e) {
            // TODO: handle exception
        }
    }
 

@Test
public void test_putItem_WithAllParameters() throws Exception {
  createTable();

  PutItemResult result = putItem(TEST_ATTRIBUTE, TEST_ATTRIBUTE_VALUE);
  Double units = result.getConsumedCapacity().getCapacityUnits();

  assertThat(units.doubleValue(), equalTo(1.0));
}
 

private PutItemResult putItem(String attributeName, String attributeValue) throws Exception {
  Map<String, AttributeValue> item = new HashMap<String, AttributeValue>();
  item.put(attributeName, new AttributeValue()
    .withS(attributeValue));
  String returnValues = "";

  PutItemResult result = dynamoDb.putItem(TEST_TABLE_NAME, item, returnValues);

  return result;
}
 

private void retryAsync(final Map<String, List<WriteRequest>> map) {
  for (final Entry<String, List<WriteRequest>> requests : map.entrySet()) {
    for (final WriteRequest r : requests.getValue()) {
      if (r.getPutRequest() != null) {

        /**
         * The code is pretty similar to retry. The only difference is retryAsync uses
         * putItemAsync instead of putItem
         */
        final PutItemRequest putRequest =
            new PutItemRequest(requests.getKey(), r.getPutRequest().getItem());
        final Future<PutItemResult> future =
            client.putItemAsync(putRequest, new AsyncHandler<PutItemRequest, PutItemResult>() {

              @Override
              public void onError(final Exception exception) {
                LOGGER.warn("Putitem Async failed in Dynamo");
                futureMap.remove(putRequest);
              }

              @Override
              public void onSuccess(final PutItemRequest request, final PutItemResult result) {
                if (futureMap.remove(request) == null) {
                  LOGGER.warn("Unable to delete PutItemRequest from futuresMap ");
                }

                return;
              }
            });

        futureMap.put(putRequest, future);
      }
    }
  }
}
 

private static void putItem(Map<String, AttributeValue> item)
{

    try
    {

        PutItemRequest putItemRequest = new PutItemRequest(TABLENAME, item);
        PutItemResult putItemResult = dynamoDBClient.putItem(putItemRequest);
        logger.info("Result: " + putItemResult);
    }
    catch (Exception e)
    {
        // TODO: handle exception
    }
}
 

/**
 * Adds a PutItem request to the transaction
 * 
 * @param request
 * @throws DuplicateRequestException if the item in the request is already involved in this transaction
 * @throws ItemNotLockedException when another transaction is confirmed to have the lock on the item in the request
 * @throws TransactionCompletedException when the transaction has already completed
 * @throws TransactionNotFoundException if the transaction does not exist
 * @throws TransactionException on unexpected errors or unresolvable OCC contention
 */
public PutItemResult putItem(PutItemRequest request) 
    throws DuplicateRequestException, ItemNotLockedException, 
        TransactionCompletedException, TransactionNotFoundException, TransactionException {
    
    PutItem wrappedRequest = new PutItem();
    wrappedRequest.setRequest(request);
    Map<String, AttributeValue> item = driveRequest(wrappedRequest);
    stripSpecialAttributes(item);
    return new PutItemResult().withAttributes(item);
}
 

@Override
public PutItemResult putItem(PutItemRequest request)
        throws AmazonServiceException, AmazonClientException {
    Map<String, ExpectedAttributeValue> expectedValues = request.getExpected();
    checkExpectedValues(request.getTableName(), Request.getKeyFromItem(request.getTableName(),
            request.getItem(), txManager), expectedValues);

    // conditional checks are handled by the above call
    request.setExpected(null);
    return txn.putItem(request);
}
 

@Override
public PutItemResult putItem(String tableName,
        Map<String, AttributeValue> item) throws AmazonServiceException,
        AmazonClientException {
    return putItem(new PutItemRequest()
            .withTableName(tableName)
            .withItem(item));
}
 

@Override
public PutItemResult putItem(String tableName,
        Map<String, AttributeValue> item, String returnValues)
        throws AmazonServiceException, AmazonClientException {
    return putItem(new PutItemRequest()
            .withTableName(tableName)
            .withItem(item)
            .withReturnValues(returnValues));
}
 

@Override
public PutItemResult putItem(PutItemRequest putItemRequest) {
	
	throw new UnsupportedOperationException();
}
 

@Override
public PutItemResult putItem(String tableName, Map<String, AttributeValue> item) {
	
	throw new UnsupportedOperationException();
}
 

@Override
public PutItemResult putItem(String tableName, Map<String, AttributeValue> item, String returnValues) {
	
	throw new UnsupportedOperationException();
}
 

@Override
public PutItemResult putItem(PutItemRequest request) {
    bh.consume(request);
    return PUT_ITEM_RESULT;
}
 

/**
 * {@inheritDoc}
 */
@SuppressWarnings("unchecked")
@Override
public List<Record> emit(final UnmodifiableBuffer<Record> buffer) {
    if (isShutdown) {
        if (buffer.getRecords().isEmpty()) {
            // This is OK, but not expected
            log.warn("Record processor called emit after calling shutdown. Continuing becuase buffer is empty.");
            return Collections.emptyList();
        } else {
            throw new IllegalStateException("Cannot emit records after emitter has been shutdown.");
        }
    }
    // Asynchronously process all writes, but block on the results.
    List<Record> records = buffer.getRecords();
    // Stores records that failed with a non-retryable exception
    final List<Record> failedRecords = Collections.synchronizedList(new ArrayList<Record>());
    // Queue of records to submit
    final BlockingQueue<Record> toSubmit = new LinkedBlockingQueue<Record>(records);
    // Used to detect when all requests have either succeeded or resulted in a non-retryable exception
    final CountDownLatch doneSignal = new CountDownLatch(records.size());
    final AtomicInteger retryCount = new AtomicInteger();
    boolean interrupted = false;
    try {
        while (doneSignal.getCount() > 0) {
            Record recordToSubmit = null;
            try {
                recordToSubmit = toSubmit.poll(WAIT_TIME_MS, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
                interrupted = true;
            }
            final Record record = recordToSubmit;
            if (null == record) {
                continue; // Check if all records have completed and if not try to poll again
            }
            // Generate the request based on the record
            AmazonWebServiceRequest request = createRequest(record);
            if (request == null) { // Should only happen if DynamoDB Streams API updates to support different operations
                                   // than {INSERT, MODIFY, REMOVE}.
                continue;
            }
            // Submit the write request based on its type
            if (request instanceof PutItemRequest) { // PUT
                getDynamodb().putItemAsync((PutItemRequest) request,
                    (AsyncHandler<PutItemRequest, PutItemResult>) getHandler(toSubmit, failedRecords, retryCount, doneSignal, record));
            } else if (request instanceof DeleteItemRequest) { // DELETE
                getDynamodb().deleteItemAsync((DeleteItemRequest) request,
                    (AsyncHandler<DeleteItemRequest, DeleteItemResult>) getHandler(toSubmit, failedRecords, retryCount, doneSignal, record));
            } else if (request instanceof UpdateItemRequest) { // UPDATE
                getDynamodb().updateItemAsync((UpdateItemRequest) request,
                    (AsyncHandler<UpdateItemRequest, UpdateItemResult>) getHandler(toSubmit, failedRecords, retryCount, doneSignal, record));
            } else { // Should only happen if DynamoDB allows a new operation other than {PutItem, DeleteItem,
                     // UpdateItem} for single item writes.
                log.warn("Unsupported DynamoDB request: " + request);
            }
        }
    } finally {
        if (interrupted) {
            Thread.currentThread().interrupt();
        }
    }
    emitCloudWatchMetrics(records, failedRecords, retryCount);
    if (!records.isEmpty()) {
        log.debug("Successfully emitted " + (records.size() - failedRecords.size()) + " records ending with sequence number "
            + buffer.getLastSequenceNumber());
    } else {
        log.debug("No records to emit");
    }
    return failedRecords;
}
 

public PutPointResult(PutItemResult putItemResult) {
	this.putItemResult = putItemResult;
}
 

public PutItemResult getPutItemResult() {
	return putItemResult;
}
 

@Override
public PutItemResult putItem(PutItemRequest request)
        throws AmazonServiceException, AmazonClientException {
    throw new UnsupportedOperationException("Use the underlying client instance instead");
}
 

@Override
public PutItemResult putItem(String tableName,
        Map<String, AttributeValue> item) throws AmazonServiceException,
        AmazonClientException {
    throw new UnsupportedOperationException("Use the underlying client instance instead");
}
 

@Override
public PutItemResult putItem(String tableName,
        Map<String, AttributeValue> item, String returnValues)
        throws AmazonServiceException, AmazonClientException {
    throw new UnsupportedOperationException("Use the underlying client instance instead");
}
 

@Override
public PutItemResult putItem(PutItemRequest request) throws AmazonServiceException, AmazonClientException {
    return getBackend().putItem(request);
}
 

@Override
public PutItemResult putItem(String tableName, Map<String, AttributeValue> item) throws AmazonServiceException, AmazonClientException {
    return getBackend().putItem(tableName, item);
}
 

@Override
public PutItemResult putItem(String tableName, Map<String, AttributeValue> item, String returnValues) throws AmazonServiceException, AmazonClientException {
    return getBackend().putItem(tableName, item, returnValues);
}