title: Why are threads using the same variable value? RxJava Mqtt
body: <p>I'm using rxmqtt (which uses rxjava and paho) to communicate whith a mqtt broker. I'm using javax to accept rest requests and publish some content to the broker and wait for a response. The code below works fine if I make one request at a time, but if I have more than one concurrent requests it only returns a response for the last one and the others fall into the timeout exception.</p>
  
  <p>The mqttConn.getMqttMessages() returns a flowable which is already subscribed to all topics i need:</p>
  
  <pre><code>public Flowable&lt;MqttMessage&gt; getMqttMessages() {
      return this.obsClient.subscribe("pahoRx/fa/#", 1);
  }
  </code></pre>
  
  <p>and MqttConnection is a singleton because i only want one single connection to broker and all the publishes are done in this connection
  I've noticed that my queryParam id is different in each thread execution of the web service request (expected behavior), but when it enters the subscription part of the code it only considers the last id value and does not pass my validation in the takeUntil method:</p>
  
  <pre><code>mqttConn.getMqttMessages().timeout(20, TimeUnit.SECONDS).takeUntil(msgRcv -&gt; {
          System.out.println("received: " + new String(msgRcv.getPayload()) + " ID: " + id);
          return id.equals(new String(msgRcv.getPayload()));
      }).blockingSubscribe(msgRcv -&gt; {
          final byte[] body = msgRcv.getPayload();
          System.out.println(new String(body)); //printing... but not sending the reponse
          response.set("Message Receiced: " + new String(msgRcv.getPayload()));
          return;
  
      }, e -&gt; {
          if (e instanceof TimeoutException) {
              response.set("Timeout Occured");
  
          } else {
              response.set("Some kind of error occured " + e.getLocalizedMessage());
          }
      });
  </code></pre>
  
  <p>The thing is, why is it only considering the last id received when each request should have its own independent thread? I've tried getting mqttConn.getMqttConnection() as a ThreadLocal object... doesn't fix.</p>
  
  <p>Full WS code: </p>
  
  <pre><code>@Path("/test")
  @GET
  public String test(@QueryParam("id") String id) throws InterruptedException, MqttException {
  
      String funcExec = "pahoRx/fe/";
      String content = "unlock with single connection to broker";
      int qos = 1;
      AtomicReference&lt;String&gt; response = new AtomicReference&lt;String&gt;();
  
      response.set("Initial Value");
      MqttConnection mqttConn = MqttConnection.getMqttConnection();
      ObservableMqttClient obsClient = mqttConn.getBrokerClient();
  
      MqttMessage msg = MqttMessage.create(78, content.getBytes(), qos, false);
      String topicPub = funcExec + id;
  
      obsClient.publish(topicPub, msg).subscribe(t -&gt; {
          System.out.println("Message Published");
      }, e -&gt; {
          System.out.println("Failed to publish message: " + e.getLocalizedMessage());
      });
  
      mqttConn.getMqttMessages().timeout(20, TimeUnit.SECONDS).takeUntil(msgRcv -&gt; {
          System.out.println("received: " + new String(msgRcv.getPayload()) + " ID: " + id);
          return id.equals(new String(msgRcv.getPayload()));
      }).blockingSubscribe(msgRcv -&gt; {
          final byte[] body = msgRcv.getPayload();
          System.out.println(new String(body)); //printing... but not sending the reponse
          response.set("Message Receiced: " + new String(msgRcv.getPayload()));
          return;
  
      }, e -&gt; {
          if (e instanceof TimeoutException) {
              response.set("Timeout Occured");
  
          } else {
              response.set("Some kind of error occured " + e.getLocalizedMessage());
          }
      });
      return response.get();
  }
  </code></pre>
  
  <p>I hope the explanation is clear enough!</p>
  
  <p>Ty in advance</p>