//二进制支持
client.publish("demo", new MqMessage("helloworld!".getBytes(StandardCharsets.UTF_8)));

//顺序消息分片支持（用于集群分散处理）
client.publish("demo", new MqMessage("helloworld!").sequence(true, "sharding-1"));

docker run -p 18602:18602 -p 8602:8602 -e folkmq.schema='sd:ws' noearorg/folkmq-server:1.2.4

const {FolkMQ} = require("@noear/folkmq");

async function main() {
    //创建客户端，并连接
    const client = await FolkMQ.createClient("folkmq:ws://127.0.0.1:18602")
                            .nameAs("demoapp")
                            .connect();

    //订阅主题，并指定加入的消费者分组
    client.subscribe("demo", null, true, message => {
        console.log(message);
    });

    //发布消息
    client.publish("demo", FolkMQ.newMqMessage("helloworld!"));
}

main();

//准备（1.取名字；2.添加事务回查）
const client = await FolkMQ.createClient("folkmq:ws://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名
    .connect();

//用于服务端发起的事务回查
client.transactionCheckback(m => {
  //极端特殊的情况下，客户端未完成事务确认。由服务端发起补尝确认
  if("1" == m.getAttr("orderId")) {
      //一般这里，需要查询数据库之类的
      m.acknowledge(true);
  }
});
    
//发送事务消息    
const tran = client.newTransaction();

try {
    client.publish("demo", FolkMQ.newMessage("demo1").attr("orderId","1").transaction(tran));
    client.publish("demo", FolkMQ.newMessage("demo2").attr("orderId","1").transaction(tran));
    client.publish("demo", FolkMQ.newMessage("demo3").attr("orderId","1").transaction(tran));
    client.publish("demo", FolkMQ.newMessage("demo4").attr("orderId","1").transaction(tran));

    tran.commit();
} catch (err) {
    tran.rollback();
}

//客户端1
const client1 = await FolkMQ.createClient("folkmq:ws://127.0.0.1:18602")
                        .nameAs("demoapp1")
                        .connect();

//客户端1监听
const router = FolkMQ.newRouter(m => m.getTag()).doOn("/hello", req => {
    req.respone(SocketD.newEntity("me to!"));
});
client1.listen(router.consume.bind(router));


//客户端2
const client2 = await FolkMQ.createClient("folkmq:ws://127.0.0.1:18602")
                        .nameAs("demoapp2")
                        .connect();

//客户端2发送
const resp = client2.send(FolkMQ.newMessage("helloworld!").tag("hello"), "demoapp1").await();
console.log(resp.dataAsString());

//准备（1.取名字；2.添加事务回查）
MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名
    .connect();

//用于服务端发起的事务回查
client.transactionCheckback(m->{
  if (m.isTransaction()) {
      //极端特殊的情况下，客户端未完成事务确认。由服务端发起补尝确认
      if("1".equals(m.getAttr("orderId"))) {
          //一般这里，需要查询数据库之类的
          m.acknowledge(true);
      }
  }
});
    
//发送事务消息    
MqTransaction tran = client.newTransaction();

try {
    client.publish("demo", new MqMessage("demo1").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo2").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo3").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo4").attr("orderId","1").transaction(tran));

    tran.commit();
} catch (Throwable e) {
    tran.rollback();
}

//客户端1
MqClient client1 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
                        .nameAs("demoapp1")
                        .connect();

//客户端1监听
client1.listen(new MqRouter(m -> m.getTag()).doOn("hello", request -> {
    System.out.println(request);
    request.acknowledge(new StringEntity("me to!"));
}));


//客户端2
MqClient client2 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
                        .nameAs("demoapp2")
                        .connect();

//客户端2发送
Reply reply = client2.send(new MqMessage("helloworld!").tag("hello"), "demoapp1").await();
System.out.println(reply.dataAsString());

//准备（1.取名字；2.添加响应实现）
MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名
    .connect();

//用于响应服务端发起的反向确认
client.response(m->{
  if (m.isTransaction()) {
      //极端特殊的情况下，客户端未完成事务确认。由服务端发起补尝确认
      if("1".equals(m.getAttr("orderId"))) {
          //一般这里，需要查询数据库之类的
          m.acknowledge(true);
      }
  }
});
    
//发送事务消息    
MqTransaction tran = client.newTransaction();

try {
    client.publish("demo", new MqMessage("demo1").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo2").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo3").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo4").attr("orderId","1").transaction(tran));

    tran.commit();
} catch (Throwable e) {
    tran.rollback();
}

//客户端2
MqClient client1 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demo-app1") 
    .connect();

//要支持 rpc 响应，要添加响应实现（MqResponseRouter 带了主体路由功能）
client1.response(new MqResponseRouter().doOn("test.hello", m -> {
    m.acknowledge(true, new StringEntity(m.getSender() +  ": me to! rev: " + m.getContent()));
}));

//客户端2
MqClient client2 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demo-app2")
    .connect();

//发起请求并等响应，同步模式
Reply reply = client2.request("demo-app1", "test.hello", new MqMessage("hello")).await();
print(reply.dataAsString());

//准备（1.取名字；2.添加响应实现）
MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名
    .connect();

//用于响应服务端发起的反向确认
client.response(m->{
  if (m.isTransaction()) {
      //极端特殊的情况下，客户端未完成事务确认。由服务端发起补尝确认
      if("1".equals(m.getAttr("orderId"))) {
          //一般这里，需要查询数据库之类的
          m.acknowledge(true);
      }
  }
});
    
//发送事务消息    
MqTransaction tran = client.newTransaction();

try {
    client.publish("demo", new MqMessage("demo1").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo2").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo3").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo4").attr("orderId","1").transaction(tran));

    tran.commit();
} catch (Throwable e) {
    tran.rollback();
}

//客户端2
MqClient client1 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demo-app1") 
    .connect();

//要支持 rpc 响应，要添加响应实现（MqResponseRouter 带了主体路由功能）
client1.response(new MqResponseRouter().doOn("test.hello", m -> {
    m.acknowledge(true, new StringEntity(m.getSender() +  ": me to! rev: " + m.getContent()));
}));

//客户端2
MqClient client2 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demo-app2")
    .connect();

//发起请求并等响应，同步模式
Reply reply = client2.request("demo-app1", "test.hello", new MqMessage("hello")).await();
print(reply.dataAsString());

//同步
client.publish("demo", new MqMessage("hello").sequence(true)));
//异步
client.publishAsync("demo", new MqMessage("hello").sequence(true)));

//消息绝对有序配置支持（发送有序，接收有序）
MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18601?ak=ak1&sk=sk1",
                "folkmq://127.0.0.1:18602?ak=ak1&sk=sk1")
        .config(c->c.sequenceMode(true).coreThreads(1).maxThreads(1))
        .connect();

//管理接口，一般用于用户自己开发管理界面，通过接口管理 FolkMQ
String json = client.call(MqApis.MQ_QUEUE_VIEW_MESSAGE, token, "demo","demoApp").get();