package me.mpj;

import java.io.*;
import java.net.Socket;
import java.util.concurrent.ExecutionException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class SocketWorker implements Runnable {
    private Socket _socket;
    private KafkaConsumer _consumer = null;
    private KafkaProducer _producer = null;

    public SocketWorker(Socket socket) {
        _socket = socket;
    }

    @Override
    public void run() {
        final InputStream inputStream;
        try {
            inputStream = _socket.getInputStream();
        } catch (IOException e) {
            System.out.println("SocketWorker error: Could not get input stream.");
            e.printStackTrace();
            return;
        }

        final InputStreamReader streamReader = new InputStreamReader(inputStream);
        BufferedReader br = new BufferedReader(streamReader);

        String line;
        try {
            // readLine blocks until line arrives or socket closes, upon which it returns null
            // readLine will be return each line up until a line feed / carriage return,
            // but will not including the line-termination character.
            while ((line = br.readLine()) != null) {

                Pattern sendPattern = Pattern.compile("send\\s(.+)\\s(.+)\\s(.+)");
                Matcher sendMatcher = sendPattern.matcher(line);

                // Handle: consume <topic> <group> <offset-reset>
                if (line.startsWith("consume")) {
                    Pattern pattern = Pattern.compile("consume\\s([a-z]+)\\s([a-z]+)\\s(smallest|largest)");
                    Matcher matcher = pattern.matcher(line);
                    if (matcher.find()) {
                        final String topic = matcher.group(1);
                        final String group = matcher.group(2);
                        final KafkaConsumer.AutoOffsetReset reset =
                                KafkaConsumer.AutoOffsetReset.valueOf(matcher.group(3));
                        _consumer = new KafkaConsumer(topic, group, reset);
                        sendLine("consume-started");
                    } else {
                        sendLine("error command-invalid THe consume command expects three parameters separated by "+
                                 "space; topic (a-z), consumer group (a-z), and offset reset ('smallest'/'largest')");
                    }
                }

                // Handle: next
                else if (line.trim().equals("next")){
                    if(_consumer != null) {
                        final String msg = new String(_consumer.stream.iterator().next().message());
                        sendLine("msg " + msg);
                    } else {
                        sendLine("error consume-not-started You need to start consuming and wait " +
                                 "for consume-started before sending next");
                    }

                }

                // Handle: commit
                else if (line.trim().equals("commit")){
                    _consumer.commitOffsets();
                    sendLine("commit-ok");
                }

                // Handle: send <partitionId> <body>
                else if (sendMatcher.find()) {
                    final String topic       = sendMatcher.group(1);
                    final String partitionId = sendMatcher.group(2);
                    final String body        = sendMatcher.group(3);
                    if(_producer == null)
                        _producer = new KafkaProducer();
                    try {
                        _producer.send(topic, partitionId, body);
                    } catch (ExecutionException e) {
                        sendLine("send-fail exception " + e.getMessage());
                    } catch (InterruptedException e) {
                        sendLine("send-fail interrupted" + e.getMessage());
                    }
                }

                else {
                    sendLine("command-invalid");
                }

            }
        } catch (IOException e) {

            if (e.getMessage() == "Connection Reset") {
                // Happens during normal operation, when clients forcibly disconnect, ignore
                return;
            }

            System.out.println("SocketWorker error: Could not read line.");
            e.printStackTrace();
        }

        // If we reach this line, it means that the socket has
        // has closed, so shut down any consumer or producer
        if (_consumer != null)
            _consumer.shutdown();
        if (_producer != null)
            _producer.shutdown();

        // Finally, ensure that socket is closed.
        try {
            if (!_socket.isClosed()) _socket.close();
        } catch (IOException e) {
            // Can't imagine us caring about this error
        }

    }

    private void sendLine(String line) {
        try {
            Boolean autoFlush = true;
            PrintWriter out = new PrintWriter(_socket.getOutputStream(), autoFlush);
            out.println(line);
        } catch (IOException e) {
            System.out.println("SocketWorker Error: Unable to send message on socket");
            e.printStackTrace();
        }

    }
}

@@ -4376,34 +4376,47 @@
 ace();%0A        %7D
-%0A%0A
+ finally %7B%0A    
         // If we
@@ -4470,16 +4470,20 @@
         
+   
+ 
 // has c
@@ -4531,24 +4531,28 @@
 cer%0A        
+    
 if (_consume
@@ -4554,32 +4554,36 @@
 nsumer != null)%0A
+    
             _con
@@ -4604,24 +4604,28 @@
 ();%0A        
+    
 if (_produce
@@ -4627,32 +4627,36 @@
 oducer != null)%0A
+    
             _pro
@@ -4682,16 +4682,20 @@
         
+ 
+   
 // Final
@@ -4728,32 +4728,36 @@
 closed.%0A        
+    
 try %7B%0A          
@@ -4754,24 +4754,28 @@
             
+  
+  
 if (!_socket
@@ -4804,32 +4804,36 @@
 lose();%0A        
+    
 %7D catch (IOExcep
@@ -4854,16 +4854,20 @@
         
+   
+ 
 // Can't
@@ -4902,33 +4902,49 @@
 s error%0A        
-%7D
+    %7D%0A        %7D%0A%0A
 %0A%0A    %7D%0A%0A    pri

clean up resources on error