Skip to content

Instantly share code, notes, and snippets.

@adamw
Created June 26, 2014 14:40
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save adamw/8eebb453e368a9bf079d to your computer and use it in GitHub Desktop.
Save adamw/8eebb453e368a9bf079d to your computer and use it in GitHub Desktop.
val serverLocator = {
val nettyParams = new java.util.HashMap[String, Object]()
nettyParams.put(TransportConstants.HOST_PROP_NAME, "localhost")
nettyParams.put(TransportConstants.PORT_PROP_NAME, "5445")
val sl = HornetQClient.createServerLocatorWithHA(
new TransportConfiguration(classOf[NettyConnectorFactory].getName, nettyParams))
sl.setConfirmationWindowSize(1048576)
sl.setBlockOnDurableSend(false)
sl.setBlockOnAcknowledge(false)
sl
}
val sf = serverLocator.createSessionFactory()
val session = sf.createSession(true, true, 0)
session.createQueue(QueueName, QueueName, true)
val producer = session.createProducer(QueueName)
val semaphore = new Semaphore(0)
session.setSendAcknowledgementHandler(new SendAcknowledgementHandler {
override def sendAcknowledged(message: Message) = {
semaphore.release(1)
}
})
def send(msgs: List[String]) {
for (rawMsg <- msgs) {
val msg = session.createMessage(true)
msg.putStringProperty(ContentPropertyName, rawMsg)
producer.send(msg)
}
semaphore.acquire(msgs.size)
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment