title: RxAndroid to work with ServerSocket - Android
body: <p>I was trying to create Server and Client application using RxAndroid,</p>
  
  <p>Since i am new to Reactive Programming, still i am confused on what Map should be used and how exactly can this be implemented?</p>
  
  <p>This is my current server using Thread:</p>
  
  <pre><code>import com.mbh.usbcom.MBLogger.MBLogger;
  
  import java.net.ServerSocket;
  
  /**
   * Created by MBH on 2016-02-01.
   */
  public class MBUSBServer {
      interface IRepliedMessageHandler {
          void repliedMessage(String message, int count);
      }
  
      interface IReceivedMessageHandler {
          void receivedMessage(String message, int count);
      }
  
      MBLogger logger;
  
      private static MBUSBServer mInstance = null;
  //    private final int PORT = 59900; // anyport
      ServerSocket serverSocket;
      private volatile boolean isRunning = false;
      private final String ReplyMessage = "Client#";
  
      IRepliedMessageHandler repliedMessageHandler;
      IReceivedMessageHandler receivedMessageHandler;
      Thread mThread;
      int clinetsCount = 0;
  
      // Singlton : 1 Server at a time
      public static MBUSBServer getInstance() {
          if (mInstance == null) {
              mInstance = new MBUSBServer();
          }
          return mInstance;
      }
  
      private MBUSBServer() {
          logger = new MBLogger.Builder()
                  .setTag(MBUSBServer.class)
                  .createLogger();
      }
  
      public void start(IReceivedMessageHandler receivedMessageHandler,
                        IRepliedMessageHandler repliedMessageHandler) {
          this.repliedMessageHandler = repliedMessageHandler;
          this.receivedMessageHandler = receivedMessageHandler;
          start();
      }
  
      private void start() {
          if (isRunning) return;
          if (mThread != null) {
              mThread = null;
          }
  
          isRunning = true;
          mThread = new Thread(new Runnable() {
              @Override
              public void run() {
                  try {
                      logger.logDebug("Initializing ServerSocket");
                      serverSocket = new ServerSocket(59900);
                      serverSocket.setReuseAddress(true);
                      logger.logDebug("ServerSocker initialized--InetAddress=" + serverSocket.getInetAddress());
  
                      while (isRunning) {
                          logger.logInfo("ServerSocket is waiting for Client");
                          MBTCPClient mbtcpClient = new MBTCPClient(serverSocket.accept());
  
                          clinetsCount++;
                          //logger.logDebug("Received Request from Client #" + clinetsCount);
  
                          // Write rely to socket, then Read from socket, and push message to handler
                          receiveAndReplyWithSocket(false, mbtcpClient, clinetsCount);
                      }
  
                      isRunning = false;
                  } catch (Exception e) {
                      isRunning = false;
                      logger.logError(e);
                  }
              }
          });
          mThread.start();
      }
  
      private void receiveAndReplyWithSocket(boolean sameThread, final MBTCPClient mbtcpClient, final int count) {
          if (sameThread) {
              receiveAndReplyWithSocket(mbtcpClient, count);
          } else {
              new Thread(new Runnable() {
                  @Override
                  public void run() {
                      receiveAndReplyWithSocket(mbtcpClient, count);
                  }
              }).start();
          }
      }
  
      private void receiveAndReplyWithSocket(MBTCPClient mbtcpClient, int count) {
          try {
  
              final String reply = ReplyMessage + count;
              mbtcpClient.write(reply);
              if (repliedMessageHandler != null) {
                  repliedMessageHandler.repliedMessage(reply, count);
              }
              String response = mbtcpClient.read();
              if (receivedMessageHandler != null) {
                  receivedMessageHandler.receivedMessage(response, count);
              }
              mbtcpClient.close();
          } catch (Exception e) {
              logger.logError(e);
          }
      }
  
      public void stop() {
          isRunning = false;
          if (mThread != null) {
              mThread = null;
          }
          if (serverSocket != null) {
              try {
                  serverSocket.close();
                  serverSocket = null;
              } catch (Exception e) {
                  e.printStackTrace();
              }
          }
          logger.logDebug("Server Has Been Stopped");
      }
  }
  </code></pre>
  
  <p>and my current client:</p>
  
  <pre><code>import com.mbh.usbcom.MBLogger.MBLogger;
  
  import java.io.BufferedReader;
  import java.io.DataOutputStream;
  import java.io.IOException;
  import java.io.InputStreamReader;
  import java.net.Socket;
  
  /**
   * Created by MBH on 2016-02-02.
   */
  public class MBTCPClient {
      Socket socket;
      DataOutputStream out;
      BufferedReader in;
      MBLogger logger = new MBLogger.Builder().setTag(MBTCPClient.class)
              .createLogger();
  
      public MBTCPClient(Socket socket) {
          this.socket = socket;
      }
  
      public MBTCPClient(String hostAdds, int port) {
          try {
              socket = new Socket(hostAdds, port);
          } catch (IOException e) {
              logger.logError(e);
          }
      }
  
      public String read() {
          try {
              in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
              return in.readLine();
          } catch (IOException e) {
              logger.logError(e);
              return null;
          }
      }
  
      public boolean write(String reply) {
          if(socket == null) return false;
          try {
              out = new DataOutputStream(socket.getOutputStream());
              out.writeBytes(reply);
              return true;
          } catch (IOException e) {
              logger.logError(e);
              return false;
          }
      }
  
      public void close() {
          // close socket
          if (socket != null) {
              try {
                  socket.close();
              } catch (IOException e) {
                  logger.logError(e);
              }
          }
  
          // close input stream
          if (in != null) {
              try {
                  in.close();
              } catch (IOException e) {
                  logger.logError(e);
              }
          }
  
          // close output stream
          if (out != null) {
              try {
                  out.close();
              } catch (IOException e) {
                  logger.logError(e);
              }
          }
      }
  
  }
  </code></pre>