public static void main(String[] args) throws InterruptedException, IOException {
        log.info("SimplePublishSubscribe");
        LocalTurtleEnvironment env = new LocalTurtleEnvironment();
        env.open();

        ConsoleReporter reporter = ConsoleReporter.forRegistry(env.getMetrics())
                .convertRatesTo(TimeUnit.SECONDS)
                .convertDurationsTo(TimeUnit.MILLISECONDS)
                .build();
        reporter.start(3, TimeUnit.SECONDS);


        final int message_to_sent = 1_000_000;
        log.info("Init done, press key to start test");
        Scanner scanner = new Scanner(System.in);
        scanner.nextLine();
        List<String> listSubscriber = new ArrayList<>();

        long startTime = System.currentTimeMillis();
        AtomicInteger messageRecvSub1 = new AtomicInteger();
        String subid = env.subscribe((header, body, firstMatchTag,sourceSubscriber) -> {
            if (messageRecvSub1.addAndGet(1) == message_to_sent) {
                long endTime = System.currentTimeMillis();
                long duration = (endTime - startTime);
                log.info("1 - **** Duration:" + duration);
            }
        }, "#pizza", "#pasta", "wine");
        listSubscriber.add(subid);


        int tag_count = 1_500_000;
        String[] p = new String[tag_count];
        for (int i = 0; i < tag_count; i++) {
            p[i] = UUID.randomUUID().toString();
        }

        AtomicInteger messageRecvSub2 = new AtomicInteger();
        String subid2 = env.subscribe((header, body, firstMatchTag,sourceSubscriber) -> {
            if (messageRecvSub2.addAndGet(1) == message_to_sent) {
                long endTime = System.currentTimeMillis();
                long duration = (endTime - startTime);
                log.info("2 - **** Duration messageRecvSub2 :" + duration);
            }
        }, p);
        listSubscriber.add(subid2);


        AtomicInteger messageRecvSub3 = new AtomicInteger();
        String subid3 = env.subscribe((header, body, firsttMatchTag,sourceSubscriber) -> {
         /*   try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }*/
            if (messageRecvSub3.addAndGet(1) == message_to_sent) {
                long endTime = System.currentTimeMillis();
                long duration = (endTime - startTime);
                log.info("3 - **** Duration messageRecvSub3 :" + duration);
            }
        }, "winef", "#spaghetti");
        listSubscriber.add(subid3);


        for (int i = 0; i < message_to_sent; i++) {
            env.publish(new String("today spaghetti and wine !!").getBytes(), "#pasta", "#wine", "#spaghetti");

        }

        long endTime = System.currentTimeMillis();
        long duration = (endTime - startTime);
        log.info(" **** Duration publish :" + duration);
        log.info("press key to de-open");
        scanner.nextLine();
        log.info("Total messages 1:" + messageRecvSub1.get());
        log.info("Total messages 2:" + messageRecvSub2.get());
        log.info("Total messages 3:" + messageRecvSub3.get());
        listSubscriber.forEach(env::unSubscribe);
        env.close();
        log.info("press key to stop");
        scanner.nextLine();


    }