sourcecode

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class Client {
	static String work_file;
	//file handlers for the work_file
			static FileReader fr;
			static BufferedReader br;
	//enlapsed time
			static	long enlapsed_time;
	public static void main(String[] args) throws Exception {
		//switch for selection of local Vs remote
		switch(args[0]){
		case "REMOTE":{
			work_file=args[1];
			fr=new FileReader(work_file);br=new BufferedReader(fr);	
			client_remote obj=new client_remote();
			break;}
		case "LOCAL":{
			work_file=args[2];
			fr=new FileReader(work_file);br=new BufferedReader(fr);
			client_local obj=new client_local(args[1]);
			break;}
		default :{
			System.out.println("pls enter the first argument as REMOTE or LOCAL");break;}
		}	
	}

}

import java.io.IOException;
import java.util.concurrent.ConcurrentLinkedQueue;

public class client_local {
	static int N;
	static worker[] w;
	//request & response queue
	static ConcurrentLinkedQueue<String> request;
	static ConcurrentLinkedQueue<String> response;
	//flag to stop worker
	static boolean run=true;
	client_local(String num_threads) throws IOException, InterruptedException{
		N=Integer.parseInt(num_threads);
		//a dynamic in memory request queue
		request=new ConcurrentLinkedQueue<String>(); 
		//a dynamic in memory response queue
		response=new ConcurrentLinkedQueue<String>();
		//Creating worker threads first
		w=new worker[N];
		for(int i=0;i<N;i++)
			w[i]=new worker();
		//creating client
		client_put c_put=new client_put();
		c_put.t.join();
		}
}
import java.io.IOException;

public class client_put implements Runnable{
	//counter for number of task found in work file & default id
	static int num_task=0;
	static Thread t;String task;
	
	client_put() throws IOException {
		t=new Thread(this);
			t.start();
	}
	@Override
	public void run(){
		// request thread
				String line;
				try {
					Client.enlapsed_time=System.currentTimeMillis();
					while((line=Client.br.readLine())!=null){
						//an auto-incrementing task_id
						num_task++;
						client_local.request.add(Integer.toString(num_task)+" "+line);
					}//checking response 
					while(client_local.response.size()!=num_task)
						continue;
					//finding time
					Client.enlapsed_time=System.currentTimeMillis()-Client.enlapsed_time;
					
					client_local.run=false;
					
					System.out.println("\nnumber of workers="+client_local.N);
					System.out.println("num_task="+client_put.num_task+"\t"+"time(sec)="+(double)Client.enlapsed_time/1000);
					System.out.println("Throughput="+(client_put.num_task*1000/Client.enlapsed_time));
					Client.br.close();
					Client.fr.close();
					//System.out.println("Client out");
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			
	}

}

import java.io.IOException;

public class client_request implements Runnable{
	//counter for number of task found in work file & default id
		static int num_task=0;
		static Thread t;String task;
		
		client_request() throws IOException {
			t=new Thread(this);
				t.start();
		}
		@Override
		public void run(){
			//Client request thread
					String line;
					try {
						Client.enlapsed_time=System.currentTimeMillis();
						while((line=Client.br.readLine())!=null){
							//an auto-incrementing task_id
							num_task++;
							//making the request
							client_remote.request.insert(line);
						}
						Client.br.close();
						Client.fr.close();
						//System.out.println("Client out");
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				
		}

}

import com.amazonaws.services.sqs.model.Message;

public class client_response implements Runnable{
	static Thread t;String task;
	int item_count=0;
	client_response(){
		t=new Thread(this);
		t.start();
	}
	@Override
	public void run() {
		while(item_count!=client_request.num_task){
			//Retrieving response (remove() retrieves at most 10 messages at a time)
			client_remote.response.remove(); 
			//checking results
			for (Message message : Q.messages){
				if(message.getBody()=="true")
					continue;
				else
					System.out.println("task failed\n");
			}
		} 
		Client.enlapsed_time=System.currentTimeMillis()-Client.enlapsed_time;
		System.out.println("num_task="+client_put.num_task+"\t"+"time(sec)="+(double)Client.enlapsed_time/1000);
		System.out.println("Throughput="+(client_put.num_task*1000/Client.enlapsed_time));
		System.out.print("purging both SQS-queues & deleting DynamoDB-table & worker");
		client_remote.request.purge();
		client_remote.response.purge();
		client_remote.t.delete();
		client_remote.w.delete_worker();
	}
}

import com.amazonaws.services.ec2.model.RunInstancesRequest;
import com.amazonaws.services.ec2.model.TerminateInstancesRequest;
import com.amazonaws.services.ec2.model.RunInstancesResult;
import com.amazonaws.services.ec2.model.DescribeInstancesRequest;
import java.awt.List;

import com.amazonaws.services.ec2.AmazonEC2Client;
/*creates workers using an ami specified*/
public class create_worker {
	
static RunInstancesRequest runInstancesRequest;
static DescribeInstancesRequest describeInstancesRequest;

	create_worker(int num_worker) {
	//configuration
	runInstancesRequest = new RunInstancesRequest();
		runInstancesRequest.withImageId("ami-f2c4229f")
        .withInstanceType("t2.micro")
        .withMinCount(num_worker)
        .withMaxCount(num_worker)
        .withKeyName("key")
        .withSecurityGroups("launch-wizard-22");
		//launch
		AmazonEC2Client amazonEC2Client=new AmazonEC2Client();
					  RunInstancesResult runInstancesResult =
				      amazonEC2Client.runInstances(runInstancesRequest);
		System.out.println("\nrequried workers have been creates"+runInstancesResult);
	}
	public void delete_worker(){
		describeInstancesRequest = new DescribeInstancesRequest(); 
		TerminateInstancesRequest terminateInstanceRequest=new TerminateInstancesRequest().withInstanceIds(describeInstancesRequest.getInstanceIds());
	}
}

import java.util.List;
import java.util.Map.Entry;

import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.sqs.AmazonSQS;
import com.amazonaws.services.sqs.AmazonSQSClient;
import com.amazonaws.services.sqs.model.CreateQueueRequest;
import com.amazonaws.services.sqs.model.CreateQueueResult;
import com.amazonaws.services.sqs.model.DeleteQueueRequest;
import com.amazonaws.services.sqs.model.GetQueueAttributesRequest;
import com.amazonaws.services.sqs.model.GetQueueAttributesResult;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.PurgeQueueRequest;

public class Q {
static String myQueueUrl ;
static AmazonSQS sqs ;
static List<Message> messages ;
	Q(){ }
	public void create(String MyQueue){

		/*
	     * The ProfileCredentialsProvider will return your [default]
	     * credential profile by reading from the credentials file located at
	     * (/home/sb/.aws/credentials).
	     */
	    AWSCredentials credentials = null;
	    try {
	        credentials = new ProfileCredentialsProvider("default").getCredentials();
	    } catch (Exception e) {
	        throw new AmazonClientException(
	                "Cannot load the credentials from the credential profiles file. " +
	                "Please make sure that your credentials file is at the correct " +
	                "location (/home/sb/.aws/credentials), and is in valid format.",
	                e);
	    }

	    sqs = new AmazonSQSClient(credentials);
	    Region usEast1 = Region.getRegion(Regions.US_EAST_1);
	    sqs.setRegion(usEast1);


	    try {
	        // Create a queue
	        System.out.println("Creating a new SQS queue called"+MyQueue+".\n");
	        CreateQueueRequest createQueueRequest = new CreateQueueRequest(MyQueue);
	        //returns url for existing once
	        myQueueUrl = sqs.createQueue(createQueueRequest).getQueueUrl();

	        // List queues
	        System.out.println("Listing all queues in your account.\n");
	        for (String queueUrl : sqs.listQueues().getQueueUrls()) {
	            System.out.println("  QueueUrl: " + queueUrl);
	        }
	        System.out.println();
	    }catch (AmazonServiceException ase) {
	        System.out.println("Caught an AmazonServiceException, which means your request made it " +
	                "to Amazon SQS, but was rejected with an error response for some reason.");
	        System.out.println("Error Message:    " + ase.getMessage());
	        System.out.println("HTTP Status Code: " + ase.getStatusCode());
	        System.out.println("AWS Error Code:   " + ase.getErrorCode());
	        System.out.println("Error Type:       " + ase.getErrorType());
	        System.out.println("Request ID:       " + ase.getRequestId());
	    } catch (AmazonClientException ace) {
	        System.out.println("Caught an AmazonClientException, which means the client encountered " +
	                "a serious internal problem while trying to communicate with SQS, such as not " +
	                "being able to access the network.");
	        System.out.println("Error Message: " + ace.getMessage());
	    }


	}
public void insert(String message){
	// Send a message
    sqs.sendMessage(new SendMessageRequest(myQueueUrl, message));

	}
public List<Message> remove(){
	// Receive messages
    ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(myQueueUrl);
    messages = sqs.receiveMessage(receiveMessageRequest).getMessages();
    for (Message message : messages) {
        System.out.println("  Message");
        System.out.println("    MessageId:     " + message.getMessageId());
        System.out.println("    ReceiptHandle: " + message.getReceiptHandle());
        System.out.println("    MD5OfBody:     " + message.getMD5OfBody());
        System.out.println("    Body:          " + message.getBody());
        for (Entry<String, String> entry : message.getAttributes().entrySet()) {
            System.out.println("  Attribute");
            System.out.println("    Name:  " + entry.getKey());
            System.out.println("    Value: " + entry.getValue());
        }
    }return messages;
}
public void delete(){
	// Delete a queue
    System.out.println("Deleting the queue.\n");
    sqs.deleteQueue(new DeleteQueueRequest(myQueueUrl));
	}
public void purge(){
	//purge queue
	System.out.println("purge queue"); 
	PurgeQueueRequest purgequeueRequest= new PurgeQueueRequest(myQueueUrl);
	sqs.purgeQueue(purgequeueRequest);
	}
}

/*
 * Copyright 2010-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 * or in the "license" file accompanying this file. This file 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.
 */

import java.util.ArrayList;
import java.util.List;

import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.ec2.AmazonEC2;
import com.amazonaws.services.ec2.AmazonEC2Client;
import com.amazonaws.services.ec2.model.CancelSpotInstanceRequestsRequest;
import com.amazonaws.services.ec2.model.DescribeSpotInstanceRequestsRequest;
import com.amazonaws.services.ec2.model.DescribeSpotInstanceRequestsResult;
import com.amazonaws.services.ec2.model.LaunchSpecification;
import com.amazonaws.services.ec2.model.RequestSpotInstancesRequest;
import com.amazonaws.services.ec2.model.RequestSpotInstancesResult;
import com.amazonaws.services.ec2.model.SpotInstanceRequest;
import com.amazonaws.services.ec2.model.TerminateInstancesRequest;

public class Requests {
    private AmazonEC2         ec2;
    private ArrayList<String> instanceIds;
    private ArrayList<String> spotInstanceRequestIds;

    /**
     * Public constructor.
     * @throws Exception
     */
    public Requests () throws Exception {
        init();
    }

    /**
     * The only information needed to create a client are security credentials
     * consisting of the AWS Access Key ID and Secret Access Key. All other
     * configuration, such as the service endpoints, are performed
     * automatically. Client parameters, such as proxies, can be specified in an
     * optional ClientConfiguration object when constructing a client.
     *
     * @see com.amazonaws.auth.BasicAWSCredentials
     * @see com.amazonaws.auth.PropertiesCredentials
     * @see com.amazonaws.ClientConfiguration
     */
    private void init() throws Exception {
        /*
         * The ProfileCredentialsProvider will return your [default]
         * credential profile by reading from the credentials file located at
         * (/home/sb/.aws/credentials).
         */
        AWSCredentials credentials = null;
        try {
            credentials = new ProfileCredentialsProvider("default").getCredentials();
        } catch (Exception e) {
            throw new AmazonClientException(
                    "Cannot load the credentials from the credential profiles file. " +
                    "Please make sure that your credentials file is at the correct " +
                    "location (/home/sb/.aws/credentials), and is in valid format.",
                    e);
        }

        ec2 = new AmazonEC2Client(credentials);
        Region usWest2 = Region.getRegion(Regions.US_WEST_2);
        ec2.setRegion(usWest2);
    }

    /**
     * The submit method will create 1 x one-time t1.micro request with a maximum bid
     * price of $0.03 using the Amazon Linux AMI.
     *
     * Note the AMI id may change after the release of this code sample, and it is important
     * to use the latest. You can find the latest version by logging into the AWS Management
     * console, and attempting to perform a launch. You will be presented with AMI options,
     * one of which will be Amazon Linux. Simply use that AMI id.
     */
    public void submitRequests() {
        //==========================================================================//
        //================= Submit a Spot Instance Request =====================//
        //==========================================================================//

        // Initializes a Spot Instance Request
        RequestSpotInstancesRequest requestRequest = new RequestSpotInstancesRequest();

        // Request 1 x t1.micro instance with a bid price of $0.03.
        requestRequest.setSpotPrice("0.03");
        requestRequest.setInstanceCount(Integer.valueOf(1));

        // Setup the specifications of the launch. This includes the instance type (e.g. t1.micro)
        // and the latest Amazon Linux AMI id available. Note, you should always use the latest
        // Amazon Linux AMI id or another of your choosing.
        LaunchSpecification launchSpecification = new LaunchSpecification();
        launchSpecification.setImageId("ami-8c1fece5");
        launchSpecification.setInstanceType("t1.micro");

        // Add the security group to the request.
        ArrayList<String> securityGroups = new ArrayList<String>();
        securityGroups.add("GettingStartedGroup");
        launchSpecification.setSecurityGroups(securityGroups);

        // Add the launch specifications to the request.
        requestRequest.setLaunchSpecification(launchSpecification);

        // Call the RequestSpotInstance API.
        RequestSpotInstancesResult requestResult = ec2.requestSpotInstances(requestRequest);
        List<SpotInstanceRequest> requestResponses = requestResult.getSpotInstanceRequests();

        // Setup an arraylist to collect all of the request ids we want to watch hit the running
        // state.
        spotInstanceRequestIds = new ArrayList<String>();

        // Add all of the request ids to the hashset, so we can determine when they hit the
        // active state.
        for (SpotInstanceRequest requestResponse : requestResponses) {
            System.out.println("Created Spot Request: "+requestResponse.getSpotInstanceRequestId());
            spotInstanceRequestIds.add(requestResponse.getSpotInstanceRequestId());
        }

    }

    /**
     * The areOpen method will determine if any of the requests that were started are still
     * in the open state. If all of them have transitioned to either active, cancelled, or
     * closed, then this will return false.
     * @return
     */
    public boolean areAnyOpen() {
        //==========================================================================//
        //============== Describe Spot Instance Requests to determine =============//
        //==========================================================================//

        // Create the describeRequest with tall of the request id to monitor (e.g. that we started).
        DescribeSpotInstanceRequestsRequest describeRequest = new DescribeSpotInstanceRequestsRequest();
        describeRequest.setSpotInstanceRequestIds(spotInstanceRequestIds);

        System.out.println("Checking to determine if Spot Bids have reached the active state...");

        // Initialize variables.
        instanceIds = new ArrayList<String>();

        try
        {
            // Retrieve all of the requests we want to monitor.
            DescribeSpotInstanceRequestsResult describeResult = ec2.describeSpotInstanceRequests(describeRequest);
            List<SpotInstanceRequest> describeResponses = describeResult.getSpotInstanceRequests();

            // Look through each request and determine if they are all in the active state.
            for (SpotInstanceRequest describeResponse : describeResponses) {
                System.out.println(" " +describeResponse.getSpotInstanceRequestId() +
                                   " is in the "+describeResponse.getState() + " state.");

                // If the state is open, it hasn't changed since we attempted to request it.
                // There is the potential for it to transition almost immediately to closed or
                // cancelled so we compare against open instead of active.
                if (describeResponse.getState().equals("open")) {
                    return true;
                }

                // Add the instance id to the list we will eventually terminate.
                instanceIds.add(describeResponse.getInstanceId());
            }
        } catch (AmazonServiceException e) {
            // Print out the error.
            System.out.println("Error when calling describeSpotInstances");
            System.out.println("Caught Exception: " + e.getMessage());
            System.out.println("Reponse Status Code: " + e.getStatusCode());
            System.out.println("Error Code: " + e.getErrorCode());
            System.out.println("Request ID: " + e.getRequestId());

            // If we have an exception, ensure we don't break out of the loop.
            // This prevents the scenario where there was blip on the wire.
            return true;
        }

        return false;
    }

    /**
     * The cleanup method will cancel and active requests and terminate any running instances
     * that were created using this object.
     */
    public void cleanup () {
        //==========================================================================//
        //================= Cancel/Terminate Your Spot Request =====================//
        //==========================================================================//
        try {
            // Cancel requests.
            System.out.println("Cancelling requests.");
            CancelSpotInstanceRequestsRequest cancelRequest = new CancelSpotInstanceRequestsRequest(spotInstanceRequestIds);
            ec2.cancelSpotInstanceRequests(cancelRequest);
        } catch (AmazonServiceException e) {
            // Write out any exceptions that may have occurred.
            System.out.println("Error cancelling instances");
            System.out.println("Caught Exception: " + e.getMessage());
            System.out.println("Reponse Status Code: " + e.getStatusCode());
            System.out.println("Error Code: " + e.getErrorCode());
            System.out.println("Request ID: " + e.getRequestId());
        }

        try {
            // Terminate instances.
            System.out.println("Terminate instances");
            TerminateInstancesRequest terminateRequest = new TerminateInstancesRequest(instanceIds);
            ec2.terminateInstances(terminateRequest);
        } catch (AmazonServiceException e) {
            // Write out any exceptions that may have occurred.
            System.out.println("Error terminating instances");
            System.out.println("Caught Exception: " + e.getMessage());
            System.out.println("Reponse Status Code: " + e.getStatusCode());
            System.out.println("Error Code: " + e.getErrorCode());
            System.out.println("Request ID: " + e.getRequestId());
        }

        // Delete all requests and instances that we have terminated.
        instanceIds.clear();
        spotInstanceRequestIds.clear();
    }

}
import java.util.HashMap;
import java.util.Map;

import com.amazonaws.AmazonClientException;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.ComparisonOperator;
import com.amazonaws.services.dynamodbv2.model.Condition;
import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.DeleteTableRequest;
import com.amazonaws.services.dynamodbv2.model.DeleteTableResult;
import com.amazonaws.services.dynamodbv2.model.DescribeTableRequest;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.PutItemRequest;
import com.amazonaws.services.dynamodbv2.model.PutItemResult;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import com.amazonaws.services.dynamodbv2.model.ScanRequest;
import com.amazonaws.services.dynamodbv2.model.ScanResult;
import com.amazonaws.services.dynamodbv2.model.TableDescription;
import com.amazonaws.services.dynamodbv2.util.Tables;

public class table {

    /*
     * Before running the code:
     *      Fill in your AWS access credentials in the provided credentials
     *      file template, and be sure to move the file to the default location
     *      (/home/sb/.aws/credentials) where the sample code will load the
     *      credentials from.
     *      https://console.aws.amazon.com/iam/home?#security_credential
     *
     * WARNING:
     *      To avoid accidental leakage of your credentials, DO NOT keep
     *      the credentials file in your source directory.
     */

    static AmazonDynamoDBClient dynamoDB;
    static String tableName;
    static Map<String,AttributeValue> item;
    /**
     * The only information needed to create a client are security credentials
     * consisting of the AWS Access Key ID and Secret Access Key. All other
     * configuration, such as the service endpoints, are performed
     * automatically. Client parameters, such as proxies, can be specified in an
     * optional ClientConfiguration object when constructing a client.
     *
     * @see com.amazonaws.auth.BasicAWSCredentials
     * @see com.amazonaws.auth.ProfilesConfigFile
     * @see com.amazonaws.ClientConfiguration
     */
    //configuration 
    private static void init() throws Exception {
        /*
         * The ProfileCredentialsProvider will return your [default]
         * credential profile by reading from the credentials file located at
         * (/home/sb/.aws/credentials).
         */
        AWSCredentials credentials = null;
        try {
            credentials = new ProfileCredentialsProvider("default").getCredentials();
        } catch (Exception e) {
            throw new AmazonClientException(
                    "Cannot load the credentials from the credential profiles file. " +
                    "Please make sure that your credentials file is at the correct " +
                    "location (/home/sb/.aws/credentials), and is in valid format.",
                    e);
        }
        dynamoDB = new AmazonDynamoDBClient(credentials);
        Region usEast1 = Region.getRegion(Regions.US_EAST_1);
        dynamoDB.setRegion(usEast1);
    }
//creating a table
public void create() throws Exception{
	init();

    try {
        tableName = "table";

        // Create table if it does not exist yet
        if (Tables.doesTableExist(dynamoDB, tableName)) {
            System.out.println("Table " + tableName + " is already ACTIVE");
        } else {
            // Create a table with a primary hash key named 'name', which holds a string
            CreateTableRequest createTableRequest = new CreateTableRequest().withTableName(tableName)
                .withKeySchema(new KeySchemaElement().withAttributeName("message").withKeyType(KeyType.HASH))
                .withAttributeDefinitions(new AttributeDefinition().withAttributeName("message").withAttributeType(ScalarAttributeType.S))
                .withProvisionedThroughput(new ProvisionedThroughput().withReadCapacityUnits(100L).withWriteCapacityUnits(100L));
                TableDescription createdTableDescription = dynamoDB.createTable(createTableRequest).getTableDescription();
            System.out.println("Created Table: " + createdTableDescription);

            // Wait for it to become active
            System.out.println("Waiting for " + tableName + " to become ACTIVE...");
            Tables.awaitTableToBecomeActive(dynamoDB, tableName);
        }

        // Describe our new table
        DescribeTableRequest describeTableRequest = new DescribeTableRequest().withTableName(tableName);
        TableDescription tableDescription = dynamoDB.describeTable(describeTableRequest).getTable();
        System.out.println("Table Description: " + tableDescription);

    }catch (AmazonServiceException ase) {
        System.out.println("Caught an AmazonServiceException, which means your request made it "
                + "to AWS, but was rejected with an error response for some reason.");
        System.out.println("Error Message:    " + ase.getMessage());
        System.out.println("HTTP Status Code: " + ase.getStatusCode());
        System.out.println("AWS Error Code:   " + ase.getErrorCode());
        System.out.println("Error Type:       " + ase.getErrorType());
        System.out.println("Request ID:       " + ase.getRequestId());
    } catch (AmazonClientException ace) {
        System.out.println("Caught an AmazonClientException, which means the client encountered "
                + "a serious internal problem while trying to communicate with AWS, "
                + "such as not being able to access the network.");
        System.out.println("Error Message: " + ace.getMessage());
    }
	}
//adding entry to the table
 public void add(String message_id){
	// Add an item
	 item=new HashMap<String,AttributeValue>();
	 item.put("message", new AttributeValue(message_id));
	 PutItemRequest putItemRequest = new PutItemRequest(tableName,item);
     PutItemResult putItemResult = dynamoDB.putItem(putItemRequest);
 }
 public boolean scan(String message_id){
	// Scan items if exist
     HashMap<String, Condition> scanFilter = new HashMap<String, Condition>();
     Condition condition = new Condition()
         .withComparisonOperator(ComparisonOperator.EQ.toString())
         .withAttributeValueList(new AttributeValue(message_id));
     scanFilter.put("message", condition);
     ScanRequest scanRequest = new ScanRequest(tableName).withScanFilter(scanFilter);
     ScanResult scanResult = dynamoDB.scan(scanRequest);
     if(scanResult != null)
    	 return true; //message already exist
     else
     {add(message_id);
    	 return false; //new message entered
     }
 }
 public void delete(){
	 //delete table
	 System.out.println("\ndeleting DynamoDb table");
	 DeleteTableRequest deleteTableRequest = new DeleteTableRequest().withTableName(tableName);
	  DeleteTableResult result = dynamoDB.deleteTable(deleteTableRequest);
 }
}

public class worker implements Runnable{
Thread t;String task;
worker(){
		t=new Thread(this);
		t.start();
}

	@Override
	public void run(){
			while(true){
			if(!client_local.request.isEmpty()){	
				task=client_local.request.remove();
			String[] element=task.split(" ");
			try {
				Thread.sleep(Integer.parseInt(element[2]));
				//response id-0 (for success)
				client_local.response.add(element[0]+"-"+Integer.toString(0));
				//System.out.println(element[2]);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				}
			}else if(client_local.run!=true)
				break;
			else{
				try {
					Thread.sleep(1);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				}
			
			}//System.out.println("worker_out");
	}

}

import java.util.Scanner;

//code to run in the remote client 
public class worker_remote {
	//can be changed using arguments
static int num_threads=1;
static Q request;static Q response;
static table t;
static worker_thread[] th;
	public static void main(String[] args) throws Exception {
		if(args[0]!=null)
			num_threads=Integer.parseInt(args[0]);
		Scanner sc=new Scanner(System.in);
		//checking for sqs & DynamoDB
		System.out.println("pls enter sqs url");
		Q.myQueueUrl=sc.nextLine();
		request=new Q();response=new Q();
		//to find existing request & response queue  
		request.create("request");response.create("response");
		System.out.println("pls enter table name");
		table.tableName=sc.nextLine();
		t=new table();
		//to give the description of existing table
		t.create(); 
		//starting threads
		for(int i=0;i<num_threads;i++)
			th[i]=new worker_thread();
		//waiting for threads to join 
		for(int i=0;i<num_threads;i++)
			th[i].t.join(); 
	}

}

mport com.amazonaws.services.sqs.model.Message;

//worker thread in remote
public class worker_thread implements Runnable{
	Thread t;String[] task; boolean duplicate;
	worker_thread(){
		t=new Thread(this);
		t.start();
	}
	@Override
	public void run() {
		while((Q.sqs.getQueueUrl("request")!=null) && (Q.sqs.getQueueUrl("response")!=null)){
			worker_remote.request.remove();
			for (Message message : Q.messages){
				//check with DynamoDb for duplicates
				duplicate=worker_remote.t.scan(message.getMessageId());
				if(duplicate==true)
					continue;
				else{
					task=message.getBody().split(" ");
					try {
						Thread.sleep(Integer.parseInt(task[1]));
					} catch (NumberFormatException | InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					//writing response
					worker_remote.response.insert("true");
				}
			}
		}
	}

}