@Test public void testSubscribe() throws IOException { DispatchChannel dispatchChannel = mock(DispatchChannel.class); dispatchManager.subscribe("foo", dispatchChannel); pubSubReplyInputStream.write(new PubSubReply(PubSubReply.Type.SUBSCRIBE, "foo", Optional.<byte[]>absent())); verify(dispatchChannel, timeout(1000)).onDispatchSubscribed(eq("foo")); }
public synchronized void subscribe(String name, DispatchChannel dispatchChannel) { Optional<DispatchChannel> previous = Optional.fromNullable(subscriptions.get(name)); subscriptions.put(name, dispatchChannel); try { pubSubConnection.subscribe(name); } catch (IOException e) { logger.warn("Subscription error", e); } if (previous.isPresent()) { dispatchUnsubscription(name, previous.get()); } }
DispatchManager extends Thread { public synchronized void subscribe(String name, DispatchChannel dispatchChannel) { Optional<DispatchChannel> previous = Optional.fromNullable(subscriptions.get(name)); subscriptions.put(name, dispatchChannel); try { pubSubConnection.subscribe(name); } catch (IOException e) { logger.warn("Subscription error", e); } if (previous.isPresent()) { dispatchUnsubscription(name, previous.get()); } } }
DispatchManager extends Thread { public synchronized void subscribe(String name, DispatchChannel dispatchChannel) { Optional<DispatchChannel> previous = Optional.fromNullable(subscriptions.get(name)); subscriptions.put(name, dispatchChannel); try { pubSubConnection.subscribe(name); } catch (IOException e) { logger.warn("Subscription error", e); } if (previous.isPresent()) { dispatchUnsubscription(name, previous.get()); } } DispatchManager(RedisPubSubConnectionFactory redisPubSubConnectionFactory,
                         Optional<DispatchChannel> deadLetterChannel); }
DispatchManager extends Thread { public synchronized void subscribe(String name, DispatchChannel dispatchChannel) { Optional<DispatchChannel> previous = Optional.fromNullable(subscriptions.get(name)); subscriptions.put(name, dispatchChannel); try { pubSubConnection.subscribe(name); } catch (IOException e) { logger.warn("Subscription error", e); } if (previous.isPresent()) { dispatchUnsubscription(name, previous.get()); } } DispatchManager(RedisPubSubConnectionFactory redisPubSubConnectionFactory,
                         Optional<DispatchChannel> deadLetterChannel); @Override void start(); void shutdown(); synchronized void subscribe(String name, DispatchChannel dispatchChannel); synchronized void unsubscribe(String name, DispatchChannel channel); boolean hasSubscription(String name); @Override void run(); }
DispatchManager extends Thread { public synchronized void subscribe(String name, DispatchChannel dispatchChannel) { Optional<DispatchChannel> previous = Optional.fromNullable(subscriptions.get(name)); subscriptions.put(name, dispatchChannel); try { pubSubConnection.subscribe(name); } catch (IOException e) { logger.warn("Subscription error", e); } if (previous.isPresent()) { dispatchUnsubscription(name, previous.get()); } } DispatchManager(RedisPubSubConnectionFactory redisPubSubConnectionFactory,
                         Optional<DispatchChannel> deadLetterChannel); @Override void start(); void shutdown(); synchronized void subscribe(String name, DispatchChannel dispatchChannel); synchronized void unsubscribe(String name, DispatchChannel channel); boolean hasSubscription(String name); @Override void run();  }