MqttHandler.jadd 10.5 KB
Newer Older
1
import java.io.IOException;
René Schöne's avatar
René Schöne committed
2
3
4
import java.util.concurrent.TimeUnit;

aspect MqttHandler {
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
public class MqttServerHandler {
  private final java.util.Map<String, MqttHandler> handlers = new java.util.HashMap<>();
  private long time;
  private java.util.concurrent.TimeUnit unit;
  private String name;

  public MqttServerHandler() {
    this("RagConnect");
  }

  public MqttServerHandler(String name) {
    this.name = name;
    setupWaitUntilReady(1, TimeUnit.SECONDS);
  }

  public void setupWaitUntilReady(long time, java.util.concurrent.TimeUnit unit) {
    this.time = time;
    this.unit = unit;
  }

  public MqttHandler resolveHandler(java.net.URI uri) throws IOException {
    MqttHandler handler = handlers.get(uri.getHost());
    if (handler == null) {
      // first connect to that server
      handler = new MqttHandler();
      if (uri.getPort() == -1) {
        handler.setHost(uri.getHost());
      } else {
        handler.setHost(uri.getHost(), uri.getPort());
      }
      handlers.put(uri.getHost(), handler);
    }
    handler.waitUntilReady(this.time, this.unit);
    return handler;
  }

  public boolean newConnection(java.net.URI uri, java.util.function.Consumer<byte[]> callback) throws IOException {
    return resolveHandler(uri).newConnection(extractTopic(uri), callback);
  }

  public void publish(java.net.URI uri, byte[] bytes) throws IOException {
    resolveHandler(uri).publish(extractTopic(uri), bytes);
  }

  public void publish(java.net.URI uri, byte[] bytes, boolean retain) throws IOException {
    resolveHandler(uri).publish(extractTopic(uri), bytes, retain);
  }

  public void publish(java.net.URI uri, byte[] bytes,
                      org.fusesource.mqtt.client.QoS qos, boolean retain) throws IOException {
    resolveHandler(uri).publish(extractTopic(uri), bytes, qos, retain);
  }

René Schöne's avatar
René Schöne committed
58
  public static String extractTopic(java.net.URI uri) {
59
60
61
62
63
64
65
66
67
68
69
70
71
72
    String path = uri.getPath();
    if (path.charAt(0) == '/') {
      path = path.substring(1);
    }
    return path;
  }

  public void close() {
    for (MqttHandler handler : handlers.values()) {
      handler.close();
    }
  }

}
73
74
75
76
77
/**
 * Helper class to receive updates via MQTT and use callbacks to handle those messages.
 *
 * @author rschoene - Initial contribution
 */
78
public class MqttHandler {
René Schöne's avatar
René Schöne committed
79
  private static final int DEFAULT_PORT = 1883;
80

René Schöne's avatar
René Schöne committed
81
  private final org.apache.logging.log4j.Logger logger;
82
83
84
  private final String name;

  /** The host running the MQTT broker. */
René Schöne's avatar
René Schöne committed
85
  private java.net.URI host;
86
  /** The connection to the MQTT broker. */
René Schöne's avatar
René Schöne committed
87
  private org.fusesource.mqtt.client.CallbackConnection connection;
88
89
  /** Whether we are connected yet */
  private final java.util.concurrent.CountDownLatch readyLatch;
René Schöne's avatar
René Schöne committed
90
  private boolean sendWelcomeMessage = true;
René Schöne's avatar
René Schöne committed
91
  private org.fusesource.mqtt.client.QoS qos;
92
  /** Dispatch knowledge */
93
  private final java.util.Map<String, java.util.List<java.util.function.Consumer<byte[]>>> callbacks;
94

95
  public MqttHandler() {
96
    this("RagConnect");
97
98
  }

99
  public MqttHandler(String name) {
René Schöne's avatar
René Schöne committed
100
    this.name = java.util.Objects.requireNonNull(name, "Name must be set");
101
    this.logger = org.apache.logging.log4j.LogManager.getLogger(MqttHandler.class);
René Schöne's avatar
René Schöne committed
102
    this.callbacks = new java.util.HashMap<>();
103
    this.readyLatch = new java.util.concurrent.CountDownLatch(1);
René Schöne's avatar
René Schöne committed
104
    this.qos = org.fusesource.mqtt.client.QoS.AT_LEAST_ONCE;
105
106
  }

107
  public MqttHandler dontSendWelcomeMessage() {
René Schöne's avatar
René Schöne committed
108
109
110
111
112
113
114
115
116
    this.sendWelcomeMessage = false;
    return this;
  }

  /**
   * Sets the host (with default port) to receive messages from, and connects to it.
   * @throws IOException if could not connect, or could not subscribe to a topic
   * @return self
   */
117
  public MqttHandler setHost(String host) throws java.io.IOException {
René Schöne's avatar
René Schöne committed
118
119
120
    return setHost(host, DEFAULT_PORT);
  }

121
122
123
124
125
  /**
   * Sets the host to receive messages from, and connects to it.
   * @throws IOException if could not connect, or could not subscribe to a topic
   * @return self
   */
126
  public MqttHandler setHost(String host, int port) throws java.io.IOException {
René Schöne's avatar
René Schöne committed
127
128
    java.util.Objects.requireNonNull(host, "Host need to be set!");

René Schöne's avatar
René Schöne committed
129
    this.host = java.net.URI.create("tcp://" + host + ":" + port);
130
131
    logger.debug("Host for {} is {}", this.name, this.host);

René Schöne's avatar
René Schöne committed
132
    org.fusesource.mqtt.client.MQTT mqtt = new org.fusesource.mqtt.client.MQTT();
133
134
    mqtt.setHost(this.host);
    connection = mqtt.callbackConnection();
René Schöne's avatar
René Schöne committed
135
    java.util.concurrent.atomic.AtomicReference<Throwable> error = new java.util.concurrent.atomic.AtomicReference<>();
136
137

    // add the listener to dispatch messages later
René Schöne's avatar
René Schöne committed
138
    connection.listener(new org.fusesource.mqtt.client.ExtendedListener() {
139
140
141
142
143
144
145
146
147
148
      public void onConnected() {
        logger.debug("Connected");
      }

      @Override
      public void onDisconnected() {
        logger.debug("Disconnected");
      }

      @Override
René Schöne's avatar
René Schöne committed
149
      public void onPublish(org.fusesource.hawtbuf.UTF8Buffer topic, org.fusesource.hawtbuf.Buffer body, org.fusesource.mqtt.client.Callback<org.fusesource.mqtt.client.Callback<Void>> ack) {
150
        String topicString = topic.toString();
151
152
        java.util.List<java.util.function.Consumer<byte[]>> callbackList = callbacks.get(topicString);
        if (callbackList == null || callbackList.isEmpty()) {
René Schöne's avatar
René Schöne committed
153
          logger.debug("Got a message, but no callback to call. Forgot to subscribe?");
154
155
156
        } else {
          byte[] message = body.toByteArray();
//          System.out.println("message = " + Arrays.toString(message));
157
158
159
          for (java.util.function.Consumer<byte[]> callback : callbackList) {
            callback.accept(message);
          }
160
161
162
163
164
        }
        ack.onSuccess(null);  // always acknowledge message
      }

      @Override
René Schöne's avatar
René Schöne committed
165
      public void onPublish(org.fusesource.hawtbuf.UTF8Buffer topicBuffer, org.fusesource.hawtbuf.Buffer body, Runnable ack) {
166
167
168
169
170
171
172
173
174
175
176
177
        logger.warn("onPublish should not be called");
      }

      @Override
      public void onFailure(Throwable cause) {
//        logger.catching(cause);
        error.set(cause);
      }
    });
    throwIf(error);

    // actually establish the connection
René Schöne's avatar
René Schöne committed
178
    connection.connect(new org.fusesource.mqtt.client.Callback<Void>() {
179
180
      @Override
      public void onSuccess(Void value) {
181
        if (MqttHandler.this.sendWelcomeMessage) {
René Schöne's avatar
René Schöne committed
182
183
184
185
186
          connection.publish("components", (name + " is connected").getBytes(), org.fusesource.mqtt.client.QoS.AT_LEAST_ONCE, false, new org.fusesource.mqtt.client.Callback<Void>() {
            @Override
            public void onSuccess(Void value) {
              logger.debug("success sending welcome message");
              setReady();
187
188
            }

René Schöne's avatar
René Schöne committed
189
190
191
192
193
194
195
196
            @Override
            public void onFailure(Throwable value) {
              logger.debug("failure sending welcome message", value);
            }
          });
        } else {
          setReady();
        }
197
198
199
200
201
202
203
204
205
206
207
208
      }

      @Override
      public void onFailure(Throwable cause) {
//        logger.error("Could not connect", cause);
        error.set(cause);
      }
    });
    throwIf(error);
    return this;
  }

René Schöne's avatar
René Schöne committed
209
  public java.net.URI getHost() {
210
211
212
    return host;
  }

René Schöne's avatar
René Schöne committed
213
  private void setReady() {
214
    readyLatch.countDown();
René Schöne's avatar
René Schöne committed
215
216
  }

René Schöne's avatar
René Schöne committed
217
  private void throwIf(java.util.concurrent.atomic.AtomicReference<Throwable> error) throws java.io.IOException {
218
    if (error.get() != null) {
René Schöne's avatar
René Schöne committed
219
      throw new java.io.IOException(error.get());
220
221
222
    }
  }

René Schöne's avatar
René Schöne committed
223
  public void setQoSForSubscription(org.fusesource.mqtt.client.QoS qos) {
224
225
226
    this.qos = qos;
  }

227
228
229
230
231
232
  public boolean newConnection(String topic, java.util.function.Consumer<byte[]> callback) {
    if (readyLatch.getCount() > 0) {
      System.err.println("Handler not ready");
      return false;
//      // should maybe be something more kind than throwing an exception here
//      throw new IllegalStateException("Updater not ready");
233
234
    }
    // register callback
235
    logger.debug("new connection for {}", topic);
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
    if (callbacks.get(topic) == null) {
      callbacks.put(topic, new java.util.ArrayList<>());

      // subscribe at broker
      org.fusesource.mqtt.client.Topic[] topicArray = { new org.fusesource.mqtt.client.Topic(topic, this.qos) };
      connection.getDispatchQueue().execute(() -> {
        connection.subscribe(topicArray, new org.fusesource.mqtt.client.Callback<byte[]>() {
          @Override
          public void onSuccess(byte[] qoses) {
            logger.debug("Subscribed to {}, qoses: {}", topic, qoses);
          }

          @Override
          public void onFailure(Throwable cause) {
            logger.error("Could not subscribe to {}", topic, cause);
          }
        });
253
      });
254
255
    }
    callbacks.get(topic).add(callback);
256
    return true;
257
258
259
260
261
262
263
264
265
266
267
  }

  /**
   * Waits until this updater is ready to receive MQTT messages.
   * If it already is ready, return immediately with the value <code>true</code>.
   * Otherwise waits for the given amount of time, and either return <code>true</code> within the timespan,
   * if it got ready, or <code>false</code> upon a timeout.
   * @param time the maximum time to wait
   * @param unit the time unit of the time argument
   * @return whether this updater is ready
   */
René Schöne's avatar
René Schöne committed
268
  public boolean waitUntilReady(long time, java.util.concurrent.TimeUnit unit) {
269
    try {
270
      return readyLatch.await(time, unit);
271
272
273
274
275
276
277
278
279
280
281
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return false;
  }

  public void close() {
    if (connection == null) {
      logger.warn("Stopping without connection. Was setHost() called?");
      return;
    }
282
283
284
285
286
287
    connection.getDispatchQueue().execute(() -> {
      connection.disconnect(new org.fusesource.mqtt.client.Callback<Void>() {
        @Override
        public void onSuccess(Void value) {
          logger.info("Disconnected {} from {}", name, host);
        }
288

289
290
291
292
293
        @Override
        public void onFailure(Throwable ignored) {
          // Disconnects never fail. And we do not care either.
        }
      });
294
295
296
297
    });
  }

  public void publish(String topic, byte[] bytes) {
298
299
300
301
302
303
304
305
    publish(topic, bytes, false);
  }

  public void publish(String topic, byte[] bytes, boolean retain) {
    publish(topic, bytes, this.qos, retain);
  }

  public void publish(String topic, byte[] bytes, org.fusesource.mqtt.client.QoS qos, boolean retain) {
306
    connection.getDispatchQueue().execute(() -> {
307
      connection.publish(topic, bytes, qos, retain, new org.fusesource.mqtt.client.Callback<Void>() {
308
309
310
311
        @Override
        public void onSuccess(Void value) {
          logger.debug("Published some bytes to {}", topic);
        }
312

313
314
        @Override
        public void onFailure(Throwable value) {
315
          logger.warn("Could not publish on topic '{}'", topic, value);
316
317
        }
      });
318
319
320
    });
  }
}
René Schöne's avatar
René Schöne committed
321
}