ntc-jnats

ntc-jnats is a module NATS java client.

Maven

<dependency>
    <groupId>com.streetcodevn</groupId>
    <artifactId>ntc-jnats</artifactId>
    <version>1.0.0</version>
</dependency>

1. Publish-Subscribe

Publisher

String subj = "msg.test";
for (int i=0; i<10; i++) {
    String msg = "hello " + i;
    NPub.getInstance("pub-notify").publish(subj, msg);
    log.info("Published PubSub["+subj+"] : '"+msg+"'");
}

Subscriber

public static class NSubscriber extends NSub {
    private final Logger log = LoggerFactory.getLogger(NSubscriber.class);

    public NSubscriber(String name) throws IOException, InterruptedException {
        super(name);
    }

    @Override
    public void execute(Message msg) {
        try {
            String data = new String(msg.getData(), StandardCharsets.UTF_8);
            log.info("NSubscriber received on PubSub ["+getSubject()+"]: '"+data+"'");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }
}

public static void main(String[] args) {
    try {
        NSubGroup nsubGroup = new NSubGroup();
        for (int i=0; i< 2; i++) {
            NSubscriber ns = new NSubscriber("sub-notify");
            nsubGroup.add(ns);
        }
        nsubGroup.start();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

2. Queue Groups

Queue Worker

public static class NWorkerEmail extends NWorker {
    private final Logger log = LoggerFactory.getLogger(NWorkerEmail.class);

    public NWorkerEmail(String name) throws IOException, InterruptedException {
        super(name);
    }

    @Override
    public void execute(Message msg) {
        try {
            String data = new String(msg.getData(), StandardCharsets.UTF_8);
            log.info("NWorkerEmail["+getGroup()+"] received on QueueWorker["+getSubject()+"]: '"+data+"'");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }
}

public static void main(String[] args) {
    try {
        NWorkerGroup workerGroup = new NWorkerGroup();
        for (int i=0; i<2; i++) {
            NWorkerEmail nw = new NWorkerEmail("worker-email");
            workerGroup.add(nw);
        }
        workerGroup.start();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

Publisher

String subj = "worker.email";
for (int i=0; i<10; i++) {
    String msg = "hello " + i;
    NPub.getInstance("pub-notify").publish(subj, msg);
    log.info("Published QueueWorker["+subj+"] : '"+msg+"'");
}

3. Request-Reply

Request

String subj = "reqres";
for (int i=0; i<10; i++) {
    String msg = "this is request " + i;
    Message resp = NReq.getInstance("req-db").publish(subj, msg);
    log.info("NReq Requested ["+subj+"] : '"+msg+"'");
    log.info("NReq Received  ["+resp.getSubject()+"] : '"+new String(resp.getData(), StandardCharsets.UTF_8)+"'");
}

Reply

public static class NResQueryDB extends NRes {
    private final Logger log = LoggerFactory.getLogger(NResQueryDB.class);
    private String reply = "this is response ==> ";

    public NResQueryDB(String name) throws IOException, InterruptedException {
        super(name);
    }

    @Override
    public void execute(Message msg) {
        try {
            String data = new String(msg.getData(), StandardCharsets.UTF_8);
            log.info("NRes["+getGroup()+"] Received on QueueNRes["+getSubject()+"]: '"+data+"'");
            String datares = reply + data;
            reply(msg, datares);
            log.info("NRes["+getGroup()+"] Reply on QueueNRes["+getSubject()+"]: '"+datares+"'");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }
}

public static void main(String[] args) {
    try {
        NResGroup resGroup = new NResGroup();
        for (int i=0; i<2; i++) {
            NResQueryDB res = new NResQueryDB("res-db");
            resGroup.add(res);
        }
        resGroup.start();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

License

This code is under the Apache License v2.