QpidConnection implements QpidStatusListener { public void sendTextMessage(Target target, Type type, String msg) throws QpidConnectionException { if (connection == null) { throw new QpidConnectionException("Message not sent: No connection to Qpid."); } if (this.exchangeMissing) { throw new QpidConnectionException("Message not sent: Qpid queue has no exchange."); } if (this.isMissingBinding) { throw new QpidConnectionException("Message not sent: Qpid queue's exchange has no bindings."); } if (this.isFlowStopped) { throw new QpidConnectionException("Message not sent: Qpid queue is FLOW_STOPPED."); } try { log.debug("Sending message to Qpid - {}:{}", target, type); Map<Type, TopicPublisher> m = this.producerMap.get(target); if (m != null) { TopicPublisher tp = m.get(type); tp.send(session.createTextMessage(msg)); } else { log.warn("Publisher did not exist! Message will not be sent!"); } } catch (JMSException jmse) { throw new QpidConnectionException("Connection issue when sending event to Qpid message bus.", jmse); } catch (Exception ex) { throw new RuntimeException("Error sending event to Qpid message bus", ex); } } @Inject  QpidConnection(QpidConfigBuilder config, Configuration candlepinConfiguration); void sendTextMessage(Target target, Type type, String msg); synchronized void connect(); Connection newConnection(); void close(); TopicSession createTopicSession(); Topic lookupTopic(String name); @Override void onStatusUpdate(QpidStatus oldStatus, QpidStatus newStatus);  }
@Test public void throwsExceptionOnMessageSendAndConnectionIsNull() { try { connection.sendTextMessage(Event.Target.CONSUMER, Event.Type.CREATED, "test"); fail("Expected a QpidConnectionException to be thrown!"); } catch (QpidConnectionException e) { assertEquals("Message not sent: No connection to Qpid.", e.getMessage()); } }