On Thu, Jun 17, 2021 at 1:48 PM Matthew Broadhead
<[email protected]> wrote:
> i added ackWait and durableName as Strings in NATSActivationSpec so it
> is working for me ok.
>
Very nice! Can you send a PR for that? I'll get it merged in. Nice work :).
> i have managed to deploy TomEE with docker fine. But when deploying
> with Eclipse WST I have to manually copy the rar into the apps folder in
> org.eclipse.wst.server.core. Is there a way to automate the process
> using the launch configuration?
>
It's been a while since I've used Eclipse, so the short answer is "I don't
know". I tended to point WST to use a TomEE install, and to actually take
control of the install (as opposed to using a folder
in org.eclipse.wst.server.core). That way you'd just set it up once.
There is one other alternative, which is to embed the .rar in the webapp
itself. I'll hack up an example this evening for this adapter, but the
general gist of it is that you'd include the chatterbox-nats-api and
chatterbox-nats-impl jars in WEB-INF/lib (along with the dependencies), and
configure the resource adapter in WEB-INF/resources.xml (which is basically
a version of tomee.xml that is local to the application). Its similar-ish
to packaging the rar in an ear, but keeps your deployable as a rar (and
should work with WST).
Jon
>
> On 17/06/2021 11:52, Matthew Broadhead wrote:
> > maybe it is possible to put the Subscription option in the
> > NATSActivationSpec
> >
> > On 17/06/2021 09:43, Matthew Broadhead wrote:
> >> Hi Jon,
> >>
> >> I was wondering how to set the SubscriptionOptions. If you look at
> >> the example below they are the third argument to the subscribe
> >> function after the MessageHandler. In my case I might want to have
> >> control over the ackWait time and switch between manualAcks and
> >> autoAcks. The durable name is quite important too.
> >> startWithLastReceived can be changed to getting all the messages or
> >> some number of messages. Sorry if I didn't mention these before
> >>
> >> streamingConnection.subscribe("scheduler:notify", new MessageHandler() {
> >> @Override
> >> public void onMessage(Message m) {
> >> ...
> >> m.ack();
> >> }
> >> }, new
> >>
> SubscriptionOptions.Builder().startWithLastReceived().manualAcks().ackWait(Duration.ofSeconds(60))
> >> .durableName("scheduler-service").build());
> >>
> >> On 16/06/2021 16:43, Jonathan Gallimore wrote:
> >>> Pushed. I've made client Id and cluster id configuration options and
> >>> added
> >>> a README. If it working for you, I'd suggest we cut a 0.3 release
> >>> unless
> >>> there are any objections.
> >>>
> >>> Jon
> >>>
> >>> On Wed, Jun 16, 2021 at 12:16 PM Jonathan Gallimore <
> >>> [email protected]> wrote:
> >>>
> >>>> Sorry, tied up with a couple of things here. Literally just opened
> >>>> the IDE
> >>>> to finish this off. :)
> >>>>
> >>>> Jon
> >>>>
> >>>> On Wed, Jun 16, 2021 at 9:46 AM Matthew Broadhead
> >>>> <[email protected]> wrote:
> >>>>
> >>>>> hi Jon,
> >>>>>
> >>>>> Is there anything else you need from my end? It feels like it is
> >>>>> nearly
> >>>>> finished. Could it pass the parameters in a similar way to how it is
> >>>>> done in the chatterbox-imap? sending them in from the tomee.xml?
> >>>>>
> >>>>>
> >>>>> On 10/06/2021 19:07, Jonathan Gallimore wrote:
> >>>>>> Thank you! That worked. I have pushed an update to my code, and I've
> >>>>> been
> >>>>>> able to send a message to NATS from a REST endpoint, and receive a
> >>>>> message
> >>>>>> from NATS via an MDB.
> >>>>>>
> >>>>>> I still need to extract the cluster ID and client ID into
> >>>>>> properties for
> >>>>>> the resource adapter, and of course, try and write up how this
> >>>>>> works.
> >>>>> Code
> >>>>>> is here:
> >>>>>> https://github.com/jgallimore/tomee-chatterbox/tree/nats, but
> >>>>> I'll
> >>>>>> merge it in once I have done these couple of changes.
> >>>>>>
> >>>>>> One other thing on my mind is that I'd like to try and find some
> >>>>>> way to
> >>>>>> make all this easier. If you've used JMS, you've used JCA, possibly
> >>>>> without
> >>>>>> realizing it, but the spec still feels very hard to get into - I
> >>>>>> wonder
> >>>>> if
> >>>>>> there is anything we can propose in that regard to try and make
> >>>>>> creating
> >>>>>> simple connectors a bit easier.
> >>>>>>
> >>>>>> Jon
> >>>>>>
> >>>>>> On Wed, Jun 9, 2021 at 4:31 PM Matthew Broadhead
> >>>>>> <[email protected]> wrote:
> >>>>>>
> >>>>>>> hi, it was my fault putting a confusing value in the docker compose
> >>>>>>> file. it should work like this
> >>>>>>>
> >>>>>>> StreamingConnectionFactory cf = new
> >>>>>>> StreamingConnectionFactory(new
> >>>>>>> Options.Builder().natsUrl("nats://localhost:4222")
> >>>>>>> .clusterId("yourclientid").clientId("anything").build());
> >>>>>>>
> >>>>>>> but you could change the docker-compose.yml to have a -cid of
> >>>>>>> yourclusterid and then do this
> >>>>>>>
> >>>>>>> StreamingConnectionFactory cf = new
> >>>>>>> StreamingConnectionFactory(new
> >>>>>>> Options.Builder().natsUrl("nats://localhost:4222")
> >>>>>>> .clusterId("yourclusterid").clientId("yourclientid").build());
> >>>>>>>
> >>>>>>> On 09/06/2021 17:15, Jonathan Gallimore wrote:
> >>>>>>>> Thanks. If I can get that test going, I can probably get the rest
> >>>>>>> working.
> >>>>>>>> I suspect there are some other bugs in there.
> >>>>>>>>
> >>>>>>>> Jon
> >>>>>>>>
> >>>>>>>> On Wed, Jun 9, 2021 at 4:04 PM Matthew Broadhead
> >>>>>>>> <[email protected]> wrote:
> >>>>>>>>
> >>>>>>>>> Thanks I will have a look now
> >>>>>>>>>
> >>>>>>>>> On 09/06/2021 15:19, Jonathan Gallimore wrote:
> >>>>>>>>>> Ok, I'm stuck. If I boot up a NATS server with your
> >>>>> docker-compose.yml
> >>>>>>>>>> file, and run the following test:
> >>>>>>>>>>
> >>>>>>>>>> @Test
> >>>>>>>>>> public void testShouldConnect() throws Exception {
> >>>>>>>>>> StreamingConnectionFactory cf = new
> >>>>>>>>>> StreamingConnectionFactory(new
> >>>>>>>>>> Options.Builder().natsUrl("nats://localhost:4222")
> >>>>>>>>>>
> >>>>>>>>> .clusterId("cluster-id").clientId("yourclientid").build());
> >>>>>>>>>> final StreamingConnection connection =
> >>>>> cf.createConnection();
> >>>>>>>>>> Assert.assertNotNull(connection);
> >>>>>>>>>>
> >>>>>>>>>> connection.close();
> >>>>>>>>>> }
> >>>>>>>>>>
> >>>>>>>>>> It fails with a timeout.
> >>>>>>>>>>
> >>>>>>>>>> I monitored the connection with wireshark, and see the following
> >>>>>>>>>>
> >>>>>>>>>> < = from server to client
> >>>>>>>>>>> = from client to server
> >>>>>>>>>> <INFO
> >>>>>>>>>>
> >>>>>
> <{"server_id":"NDMRYDSGUSH2QR6SZWMFB44ND5CODXGKNYTQ5IPLUGYUDBI6G54CIGF6","server_name":"NDMRYDSGUSH2QR6SZWMFB44ND5CODXGKNYTQ5IPLUGYUDBI6G54CIGF6","version":"2.1.4","proto":1,"git_commit":"fb009af","go":"go1.13.7","host":"0.0.0.0","port":4222,"max_payload":1048576,"client_id":10}
>
> >>>>>
> >>>>>>>>>>> CONNECT
> >>>>>
> {"lang":"java","version":"2.6.5","name":"yourclientid","protocol":1,"verbose":false,"pedantic":false,"tls_required":false,"echo":true}
>
> >>>>>
> >>>>>>>>>>> PING
> >>>>>>>>>> <PONG
> >>>>>>>>>>> SUB _STAN.acks.LP4bdY88abuVJ19Qo5HVuk 1
> >>>>>>>>>>> SUB _INBOX.LP4bdY88abuVJ19Qo5HVn8 2
> >>>>>>>>>>> SUB _INBOX.LP4bdY88abuVJ19Qo5HVqw 3
> >>>>>>>>>>> SUB _INBOX.F0vVy1N0sQM3xseeEWMIAL.* 4
> >>>>>>>>>>> PUB _STAN.discover.cluster-id
> >>>>>>>>>> _INBOX.F0vVy1N0sQM3xseeEWMIAL.F0vVy1N0sQM3xseeEWMISH 75
> >>>>>
> .yourclientid.._INBOX.LP4bdY88abuVJ19Qo5HVn8..".LP4bdY88abuVJ19Qo5HVjK(.0.
> >>>>>
> >>>>>>>>>> <PING
> >>>>>>>>>>> PONG
> >>>>>>>>>>> UNSUB 1
> >>>>>>>>>>> UNSUB 2
> >>>>>>>>>>> UNSUB 3
> >>>>>>>>>> So there does appear to be some communication between my test
> >>>>>>>>>> and
> >>>>> the
> >>>>>>>>> NATS
> >>>>>>>>>> server - I have no idea why it times out.
> >>>>>>>>>>
> >>>>>>>>>> My code is here if you want to have a go:
> >>>>>>>>>> https://github.com/jgallimore/tomee-chatterbox/tree/nats
> >>>>>>>>>>
> >>>>>>>>>> Jon
> >>>>>>>>>>
> >>>>>>>>>> On Wed, Jun 9, 2021 at 11:56 AM Jonathan Gallimore <
> >>>>>>>>>> [email protected]> wrote:
> >>>>>>>>>>
> >>>>>>>>>>> Nervermind, I figured out my mistake. I'll post back when I
> >>>>>>>>>>> have
> >>>>>>>>> something
> >>>>>>>>>>> going.
> >>>>>>>>>>>
> >>>>>>>>>>> Jon
> >>>>>>>>>>>
> >>>>>>>>>>> On Wed, Jun 9, 2021 at 11:44 AM Jonathan Gallimore <
> >>>>>>>>>>> [email protected]> wrote:
> >>>>>>>>>>>
> >>>>>>>>>>>> I think I have something wired up, but when executing this:
> >>>>>>>>>>>>
> >>>>>>>>>>>> cf = new
> >>>>>>>>>>>> StreamingConnectionFactory(new
> >>>>>>>>>>>> Options.Builder().natsUrl(baseAddress)
> >>>>>>>>>>>>
> >>>>>>>>>>>> .clusterId("cluster-id").clientId("client-id").build());
> >>>>>>>>>>>>
> >>>>>>>>>>>> connection = cf.createConnection();
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>> connection is null. Any pointers?
> >>>>>>>>>>>>
> >>>>>>>>>>>> Jon
> >>>>>>>>>>>>
> >>>>>>>>>>>> On Wed, Jun 9, 2021 at 8:16 AM Matthew Broadhead
> >>>>>>>>>>>> <[email protected]> wrote:
> >>>>>>>>>>>>
> >>>>>>>>>>>>> i have never used a JCA adapter before. is it loaded in
> >>>>>>>>>>>>> using
> >>>>> the
> >>>>>>>>>>>>> tomee.xml as a Resource? and then injected into a
> >>>>>>>>>>>>> singleton for
> >>>>>>>>>>>>> subscribing to messages?
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> On 08/06/2021 17:15, Jonathan Gallimore wrote:
> >>>>>>>>>>>>>> Definitely sounds like a good case for a JCA adapter.
> >>>>>>>>>>>>>> I'll take
> >>>>> a
> >>>>>>>>> quick
> >>>>>>>>>>>>>> swing at hooking up an example for you.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Jon
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> On Tue, Jun 8, 2021 at 9:02 AM Matthew Broadhead
> >>>>>>>>>>>>>> <[email protected]> wrote:
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Hi Jon,
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> NATS is basically a message queue, like ActiveMQ I suppose.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> I included the adapter into the project using maven
> >>>>>>>>>>>>>>> <dependency>
> >>>>>>>>>>>>>>> <groupId>io.nats</groupId>
> >>>>>>>>>>>>>>> <artifactId>java-nats-streaming</artifactId>
> >>>>>>>>>>>>>>> <version>2.2.3</version>
> >>>>>>>>>>>>>>> </dependency>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> i started up a nats server using docker. here is my
> >>>>>>>>>>>>> docker-compose.yml
> >>>>>>>>>>>>>>> version: '3.1'
> >>>>>>>>>>>>>>> services:
> >>>>>>>>>>>>>>> nats-docker:
> >>>>>>>>>>>>>>> image: nats-streaming:0.17.0
> >>>>>>>>>>>>>>> restart: always
> >>>>>>>>>>>>>>> command:
> >>>>>>>>>>>>>>> - '-p'
> >>>>>>>>>>>>>>> - '4222'
> >>>>>>>>>>>>>>> - '-m'
> >>>>>>>>>>>>>>> - '8222'
> >>>>>>>>>>>>>>> - '-hbi'
> >>>>>>>>>>>>>>> - '5s'
> >>>>>>>>>>>>>>> - '-hbt'
> >>>>>>>>>>>>>>> - '5s'
> >>>>>>>>>>>>>>> - '-hbf'
> >>>>>>>>>>>>>>> - '2'
> >>>>>>>>>>>>>>> - '-SD'
> >>>>>>>>>>>>>>> - '-cid'
> >>>>>>>>>>>>>>> - 'yourclientid'
> >>>>>>>>>>>>>>> environment:
> >>>>>>>>>>>>>>> TZ: Europe/London
> >>>>>>>>>>>>>>> LANG: en_GB.UTF-8
> >>>>>>>>>>>>>>> LANGUAGE: en_GB:en
> >>>>>>>>>>>>>>> LC_ALL: en_GB.UTF-8
> >>>>>>>>>>>>>>> ports:
> >>>>>>>>>>>>>>> - '4222:4222'
> >>>>>>>>>>>>>>> - '8222:8222'
> >>>>>>>>>>>>>>> expose:
> >>>>>>>>>>>>>>> - 4222
> >>>>>>>>>>>>>>> - 8222
> >>>>>>>>>>>>>>> networks:
> >>>>>>>>>>>>>>> - backend
> >>>>>>>>>>>>>>> networks:
> >>>>>>>>>>>>>>> backend:
> >>>>>>>>>>>>>>> driver: bridge
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> JCA sounds good if it solves the threading issue. it is
> >>>>>>>>>>>>>>> very
> >>>>> kind
> >>>>>>>>> of
> >>>>>>>>>>>>>>> you to offer to help write an adapter. looking at the
> >>>>>>>>>>>>>>> code you
> >>>>>>> sent
> >>>>>>>>>>>>> it
> >>>>>>>>>>>>>>> looks complicated but i can have a stab at it if you
> >>>>>>>>>>>>>>> don't have
> >>>>>>> much
> >>>>>>>>>>>>> time
> >>>>>>>>>>>>>>> let me know if you need more info
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Matthew
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> On 07/06/2021 17:48, Jonathan Gallimore wrote:
> >>>>>>>>>>>>>>>> At the risk of sounding a bit ignorant... what is NATS?
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> From what I can tell, it sounds like you're
> >>>>>>>>>>>>>>>> receiving a
> >>>>>>> stream
> >>>>>>>>> of
> >>>>>>>>>>>>> events
> >>>>>>>>>>>>>>>> (over websocket) and want to do some processing in an
> >>>>>>>>>>>>>>>> EJB or
> >>>>> CDI
> >>>>>>>>>>>>> bean for
> >>>>>>>>>>>>>>>> each event. The connection to the NATS server isn't in the
> >>>>>>> context
> >>>>>>>>>>>>> of a
> >>>>>>>>>>>>>>>> HTTP (or any other type of) request, and just runs all the
> >>>>> time
> >>>>>>>>>>>>> while the
> >>>>>>>>>>>>>>>> server is running - does that sound about right?
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Assuming that sounds right, it sounds a bit like the
> >>>>>>>>>>>>>>>> Slack JCA
> >>>>>>>>>>>>> connector
> >>>>>>>>>>>>>>> I
> >>>>>>>>>>>>>>>> wrote a while back:
> >>>>>>>>>>>>>>>>
> >>>>>
> https://github.com/apache/tomee-chatterbox/tree/master/chatterbox-slack
> >>>>>
> >>>>>>> .
> >>>>>>>>>>>>>>>> Essentially, the resource adapter connects to slack and
> >>>>>>>>>>>>>>>> runs
> >>>>> all
> >>>>>>>>> the
> >>>>>>>>>>>>>>> time.
> >>>>>>>>>>>>>>>> Messages that come into the server from slack are
> >>>>>>>>>>>>>>>> processed in
> >>>>>>> MDBs
> >>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>> implement the InboundListener interface.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> JCA certainly feels complex, especially when compared with
> >>>>> your
> >>>>>>>>>>>>>>>> Singleton @Startup bean approach, but I usually find
> >>>>>>>>>>>>>>>> that if I
> >>>>>>> try
> >>>>>>>>>>>>> and
> >>>>>>>>>>>>>>> work
> >>>>>>>>>>>>>>>> with threads in EJBs, things usually go in the wrong
> >>>>> direction.
> >>>>>>>>>>>>>>> Conversely,
> >>>>>>>>>>>>>>>> JCA even gives you a work manager to potentially handle
> >>>>>>>>>>>>>>>> that
> >>>>>>> stuff.
> >>>>>>>>>>>>>>>> If you can give me some pointers to running a NATS server,
> >>>>> I'd be
> >>>>>>>>>>>>> happy
> >>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>> help with a sample adapter and application.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Jon
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> On Mon, Jun 7, 2021 at 11:49 AM Matthew Broadhead
> >>>>>>>>>>>>>>>> <[email protected]> wrote:
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> I am trying to subscribe to a NATS streaming server with
> >>>>>>>>>>>>>>>>> https://github.com/nats-io/stan.java which is
> >>>>>>>>>>>>> java.lang.Autocloseable.
> >>>>>>>>>>>>>>>>> At first it wasn't closing properly as seen in my
> >>>>>>>>>>>>>>>>> original
> >>>>> gist:
> >>>>>>> https://gist.github.com/chongma/2a3ab451f2aeabc98340a9b897394cfe
> >>>>>>>>>>>>>>>>> This was solved with this
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>
> >>>>>
> https://stackoverflow.com/questions/39080296/hazelcast-threads-prevent-tomee-from-stopping
> >>>>>
> >>>>>>>>>>>>>>>>> creating a default producer:
> >>>>>>>>>>>>>>>>> @ApplicationScoped
> >>>>>>>>>>>>>>>>> public class NatsConnectionProducer {
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> @Resource(name = "baseAddressNats")
> >>>>>>>>>>>>>>>>> private String baseAddressNats;
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> @Produces
> >>>>>>>>>>>>>>>>> @ApplicationScoped
> >>>>>>>>>>>>>>>>> public StreamingConnection instance() throws
> >>>>>>> IOException,
> >>>>>>>>>>>>>>>>> InterruptedException {
> >>>>>>>>>>>>>>>>> StreamingConnectionFactory cf = new
> >>>>>>>>>>>>>>>>> StreamingConnectionFactory(new
> >>>>>>>>>>>>>>> Options.Builder().natsUrl(baseAddressNats)
> >>>>>>>>>>>>>>>>> .clusterId("cluster-id").clientId("client-id").build());
> >>>>>>>>>>>>>>>>> return cf.createConnection();
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> public void destroy(@Disposes final
> >>>>> StreamingConnection
> >>>>>>>>>>>>> instance)
> >>>>>>>>>>>>>>>>> throws IOException, TimeoutException,
> >>>>>>>>>>>>>>> InterruptedException {
> >>>>>>>>>>>>>>>>> instance.close();
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> But now i am creating a new thread because any injections
> >>>>> with
> >>>>>>> JPA
> >>>>>>>>>>>>> had
> >>>>>>>>>>>>>>>>> cacheing issues and this seems to work but i am not
> >>>>>>>>>>>>>>>>> sure it
> >>>>> is
> >>>>>>>>>>>>>>>>> broadcasting to websockets correctly
> >>>>>>>>>>>>>>>>> @Singleton
> >>>>>>>>>>>>>>>>> @Lock(LockType.READ)
> >>>>>>>>>>>>>>>>> @Startup
> >>>>>>>>>>>>>>>>> public class SchedulerEvents {
> >>>>>>>>>>>>>>>>> private static final Logger log =
> >>>>>>>>>>>>>>>>> Logger.getLogger(SchedulerEvents.class.getName());
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> @Inject
> >>>>>>>>>>>>>>>>> private StreamingConnection
> >>>>>>>>>>>>>>>>> streamingConnection;
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> @Inject
> >>>>>>>>>>>>>>>>> private SomeController someController;
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> @PostConstruct
> >>>>>>>>>>>>>>>>> private void construct() {
> >>>>>>>>>>>>>>>>> // log.fine(Thread.currentThread().getName());
> >>>>>>>>>>>>>>>>> try {
> >>>>>>>>>>>>>>>>>
> >>>>>>> streamingConnection.subscribe("scheduler:notify",
> >>>>>>>>> new
> >>>>>>>>>>>>>>>>> MessageHandler() {
> >>>>>>>>>>>>>>>>> @Override
> >>>>>>>>>>>>>>>>> public void onMessage(Message m) {
> >>>>>>>>>>>>>>>>> try {
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>> log.fine(Thread.currentThread().getName());
> >>>>>>>>>>>>>>>>> // this needs to spawn a new
> >>>>> thread
> >>>>>>>>>>>>> otherwise
> >>>>>>>>>>>>>>>>> injections are stale
> >>>>>>>>>>>>>>>>> Thread thread = new Thread(new
> >>>>>>>>>>>>> Runnable() {
> >>>>>>>>>>>>>>>>> public void run() {
> >>>>>>>>>>>>>>>>> log.fine(Thread.currentThread().getName());
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> process(m.getData());
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>> });
> >>>>>>>>>>>>>>>>> thread.start();
> >>>>>>>>>>>>>>>>> while (thread.isAlive()) {
> >>>>>>>>>>>>>>>>> // wait
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>> log.fine("Thread finished OK");
> >>>>>>>>>>>>>>>>> m.ack();
> >>>>>>>>>>>>>>>>> } catch (Exception e) {
> >>>>>>>>>>>>>>>>>
> >>>>> emailController.emailStackTrace(e);
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>> }, new
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>
> >>>>>
> SubscriptionOptions.Builder().startWithLastReceived().manualAcks().ackWait(Duration.ofSeconds(60))
>
> >>>>>
> >>>>>>>>> .durableName("scheduler-service").build());
> >>>>>>>>>>>>>>>>> } catch (IOException | InterruptedException |
> >>>>>>>>>>>>>>> TimeoutException e)
> >>>>>>>>>>>>>>>>> {
> >>>>>>>>>>>>>>>>> e.printStackTrace();
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> private void process(byte[] data) {
> >>>>>>>>>>>>>>>>> String raw = new String(data);
> >>>>>>>>>>>>>>>>> JsonReader jsonReader =
> >>>>>>>>>>>>>>>>> Json.createReader(new
> >>>>>>>>>>>>>>> StringReader(raw));
> >>>>>>>>>>>>>>>>> JsonObject jo = jsonReader.readObject();
> >>>>>>>>>>>>>>>>> jsonReader.close();
> >>>>>>>>>>>>>>>>> String type =
> >>>>>>>>>>>>>>>>> utilityDao.readJsonString(jo,
> >>>>>>> "type");
> >>>>>>>>>>>>>>>>> int id = utilityDao.readJsonInteger(jo, "id");
> >>>>>>>>>>>>>>>>> if (type == null || id == 0) {
> >>>>>>>>>>>>>>>>> emailController.emailThrowable(new
> >>>>> Throwable(),
> >>>>>>>>> raw);
> >>>>>>>>>>>>>>>>> return;
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>> log.info("Received a message: id: " +
> >>>>>>>>>>>>>>>>> id + ",
> >>>>>>> type:"
> >>>>>>>>> +
> >>>>>>>>>>>>> type);
> >>>>>>>>>>>>>>>>> DefaultServerEndpointConfigurator dsec = new
> >>>>>>>>>>>>>>>>> DefaultServerEndpointConfigurator();
> >>>>>>>>>>>>>>>>> SomeWebSocket nws =
> >>>>>>>>>>>>>>> dsec.getEndpointInstance(SomeWebSocket.class);
> >>>>>>>>>>>>>>>>> nws.broadcast(ja.toString());
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> what is the best way to use an autocloseable?
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>
> >>>>>
> >>
> >>
> >
>
>
