title: rxjava Why are not all the subscribers picking the event?
body: <p>I have been working on building a system where I watch on a socket for messages and send out to the subscribers. The subscribers will then process the message if its for them, that can be known from the message sent to them. I believe this can be achieved using rxjava. </p>
  
  <p>The problem I'm facing is, say, </p>
  
  <p>I have two subscribers, and for message number 123, only one of them are notified <code>not both</code>. 
  I'm a rxjava/reactive programming newbie.</p>
  
  <p><strong>CODE :</strong> </p>
  
  <p>I have clients connected using </p>
  
  <pre><code>public void getClientConnections(ServerSocket clientSocket, Flowable&lt;EventData&gt; events) {
      while (true) {
          try {
              Socket socket = clientSocket.accept();
          Observable.just(socket).subscribeOn(scheduler).subscribe((x) -&gt; {
                  String userId = readValueFromInputStream(socket);
                  UserData ud = new UserData(socket);
                  ud.userId = userId;
  events.observeOn(scheduler).subscribeOn(scheduler).subscribe((event) -&gt; {
                      processMessageAtClient(ud, event);
                  });
              });
          } catch (IOException e) {
              e.printStackTrace();
          }
      }
  }
  </code></pre>
  
  <p>The <code>Flowable&lt;EventData&gt; events</code> is from below</p>
  
  <pre><code>public Flowable&lt;EventData&gt; getEvents(Socket eventSocket) throws IOException {
      Flowable&lt;EventData&gt; flowable = Flowable.create((source) -&gt; {
          BufferedReader in = new BufferedReader(new InputStreamReader(eventSocket.getInputStream()));
          while (!source.isCancelled()) {
  
              String inputLine = in.readLine();
              if (inputLine != null &amp;&amp; !inputLine.isEmpty()) {
                  source.onNext(new EventData(inputLine));
              }
              // emitter.onError(new Throwable("SOMETHING WRONG"));
          }
      }, BackpressureStrategy.ERROR).map(model -&gt; {
          processInputLine((EventData) model);
          return (EventData) model;
      }).filter(model -&gt; {
          return (model.eventType != null);
      });
      return flowable;
  }
  </code></pre>
  
  <p>The Thread details are </p>
  
  <pre><code>ExecutorService threadPoolExecutor = Executors.newFixedThreadPool(100);
  Scheduler scheduler = Schedulers.from(threadPoolExecutor);
  </code></pre>
  
  <p><strong>UPDATE</strong></p>
  
  <pre><code>private void processMessageAtClient(UserData ud, EventData event) {
  
      Socket socket = ud.socket;
      String userId = ud.userId;
      if (event.toUserId.equals(userId))) {
          PrintWriter out;
          try {
              out = new PrintWriter(socket.getOutputStream(), true);
              sortedEvents.poll();
              incrementCount();
              out.println(event.inputLine);
              System.out.println("Sending " + event.inputLine);
          } catch (IOException e) {
              e.printStackTrace();
          }
      }
  }
  </code></pre>