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
<matthew.broadh...@nbmlaw.co.uk.invalid> 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
> > <matthew.broadh...@nbmlaw.co.uk.invalid> 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
> >>> <matthew.broadh...@nbmlaw.co.uk.invalid> 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?
> >>>>
> >>>>
> >>
> >>
>
>
