package com.scopely.infrastructure.kinesis;

import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.kinesis.connectors.KinesisConnectorConfiguration;
import com.amazonaws.services.kinesis.connectors.KinesisConnectorExecutorBase;
import com.amazonaws.services.kinesis.connectors.KinesisConnectorRecordProcessorFactory;
import com.amazonaws.services.s3.AmazonS3;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Locale;
import java.util.Properties;

public class KinesisRecorder extends KinesisConnectorExecutorBase<byte[], byte[]> {
    private static final Logger LOGGER = LoggerFactory.getLogger(KinesisRecorder.class);

    private final KinesisConnectorConfiguration connectorConfiguration;
    private final AmazonS3 s3;

    public KinesisRecorder(VcrConfiguration vcrConfiguration, AmazonS3 s3, AWSCredentialsProvider credentialsProvider) {
        this.s3 = s3;

        Properties properties = new Properties();
        properties.setProperty(KinesisConnectorConfiguration.PROP_APP_NAME,
                String.format(Locale.US, "kinesis-recorder-%s", vcrConfiguration.sourceStream));
        properties.setProperty(KinesisConnectorConfiguration.PROP_KINESIS_INPUT_STREAM, vcrConfiguration.sourceStream);
        properties.setProperty(KinesisConnectorConfiguration.PROP_S3_BUCKET, vcrConfiguration.bucket);
        properties.setProperty(KinesisConnectorConfiguration.PROP_BUFFER_BYTE_SIZE_LIMIT,
                String.valueOf(vcrConfiguration.bufferSizeBytes));
        properties.setProperty(KinesisConnectorConfiguration.PROP_BUFFER_MILLISECONDS_LIMIT,
                String.valueOf(vcrConfiguration.bufferTimeMillis));

        // Check everything
        if (!s3.doesBucketExist(vcrConfiguration.bucket)) {
            throw new IllegalArgumentException("Requested bucket " + vcrConfiguration.bucket + " does not exist.");
        }

        connectorConfiguration = new KinesisConnectorConfiguration(properties, credentialsProvider);
        initialize(connectorConfiguration);
    }

    @Override
    public KinesisConnectorRecordProcessorFactory<byte[], byte[]> getKinesisConnectorRecordProcessorFactory() {
        return new KinesisConnectorRecordProcessorFactory<>(new S3RecorderPipeline(s3), connectorConfiguration);
    }
}

@@ -437,16 +437,83 @@
 ctory;%0A%0A
+import java.net.InetAddress;%0Aimport java.net.UnknownHostException;%0A
 import j
@@ -529,16 +529,16 @@
 Locale;%0A
-
 import j
@@ -557,16 +557,39 @@
 perties;
+%0Aimport java.util.UUID;
 %0A%0Apublic
@@ -1767,16 +1767,112 @@
 illis));
+%0A        properties.setProperty(KinesisConnectorConfiguration.PROP_WORKER_ID, createWorkedId());
 %0A%0A      
@@ -2354,24 +2354,24 @@
 Factory() %7B%0A
-
         retu
@@ -2476,11 +2476,423 @@
 );%0A    %7D
+%0A%0A    protected static String createWorkedId() %7B%0A        String workerId = String.format(%22localhost:%25s%22, UUID.randomUUID());%0A        try %7B%0A            workerId = InetAddress.getLocalHost().getCanonicalHostName() + %22:%22 + UUID.randomUUID();%0A        %7D catch (UnknownHostException e) %7B%0A            LOGGER.warn(%22Couldn't generate proper worker id - couldn't resolve host%22, e);%0A        %7D%0A        return workerId;%0A    %7D
 %0A%7D%0A

Include worker IP in worker ID, so we don't default to VMID