package com.elab.redis.redisson.doc;

import com.elab.core.utils.RandomUtils;
import com.elab.redis.RedisSpringBoot;
import com.elab.redis.redisson.doc.model.AnyObject;
import com.elab.redis.redisson.doc.model.BloomFilterObject;
import com.elab.redis.redisson.doc.model.SomeObject;
import org.junit.Test;
import org.redisson.api.*;
import org.redisson.api.listener.MessageListener;

import java.util.concurrent.CountDownLatch;

/**
 * 分布式对象
 *
 * @author ： liukx
 * @time ： 2020/7/9 - 14:10
 */
public class DistributedTest extends RedisSpringBoot {

    private void log(int index, Object text) {
        System.out.println("-----[" + index + "]----->" + text);
    }

    @Test
    public void bigBucket() {
        RBucket<AnyObject> bucket = client.getBucket("anyObject");

        bucket.set(new AnyObject(1));
        AnyObject obj = bucket.get();
        log(1, obj);
        boolean b = bucket.trySet(new AnyObject(3));
        log(2, b);
        boolean b1 = bucket.compareAndSet(new AnyObject(4), new AnyObject(5));
        log(3, b1);
        AnyObject andSet = bucket.getAndSet(new AnyObject(6));
        log(4, andSet);
    }

    @Test
    public void bitSetTest() {
        RBitSet set = client.getBitSet("simpleBitset");
        set.set(0, true);
        set.set(1812, false);
        set.clear(0);
        set.xor("anotherBitset");
    }

    @Test
    public void atomicTest() {
        RAtomicLong atomicLong = client.getAtomicLong("myAtomicLong");
        atomicLong.set(3);
        atomicLong.incrementAndGet();
        atomicLong.get();
    }

    @Test
    public void pull() throws Exception {
        // 支持正则表达式
//        RTopic topic = client.getTopic("anyTopic*");
        RTopic topic = client.getTopic("anyTopic");
        topic.addListener(new MessageListener<SomeObject>() {
            @Override
            public void onMessage(CharSequence charSequence, SomeObject someObject) {
                log(2, someObject.toString());
            }
        });
        log(1, "开始准备就绪");
        System.in.read();
    }

    @Test
    public void push() throws Exception {
        for (int i = 0; i < 5; i++) {


            run(10, () -> {
                // in other thread or JVM
                RTopic pushTopic = client.getTopic("anyTopic");
                long clientsReceivedMessage = pushTopic.publish(new SomeObject("abc_" + RandomUtils.randomString(5)));
                log(1, clientsReceivedMessage);
            });
            Thread.sleep(5000);
            log(i, "准备发送");
        }
    }


    @Test
    public void bloomFilterTest() {
        RBloomFilter<BloomFilterObject> bloomFilter = client.getBloomFilter("sample");
        // initialize bloom filter with
        // expectedInsertions = 55000000
        // falseProbability = 0.03
        // 使用布隆过滤器之前必须先设置好容量大小和区分度
        bloomFilter.tryInit(55000000L, 0.03);

        bloomFilter.add(new BloomFilterObject("field1Value", "field2Value"));
        bloomFilter.add(new BloomFilterObject("field5Value", "field8Value"));

        boolean contains = bloomFilter.contains(new BloomFilterObject("field1Value", "field2Value"));
        log(1, contains);
        long count = bloomFilter.count();
        log(2, count);
    }

    @Test
    public void hyperLogTest() {
        // 集群模式下 需要带前缀,表示落在同一个节点上
        String prefix = "{xxx}_";
        RHyperLogLog<Integer> log = client.getHyperLogLog(prefix + "log");
        log.add(1);
        log.add(2);
        log.add(4);

        // 添加进去判断是否存在
        log(1, log.add(3));

        // 判断去重之后得到的总数，如果是不同节点可能会统计报错
        log(2, log.countWith(prefix + "log1"));

        log(5, log.count());
    }


    @Test
    public void rateLimiterTest() {
        RRateLimiter rateLimiter = client.getRateLimiter("myRateLimiter");
// 初始化
// 最大流速 = 每1秒钟产生10个令牌
        rateLimiter.trySetRate(RateType.OVERALL, 10, 1, RateIntervalUnit.SECONDS);

        CountDownLatch latch = new CountDownLatch(2);
        rateLimiter.acquire(3);
// ...

        Thread t = new Thread(() -> {
            rateLimiter.acquire(2);
            // ...
        });
    }


}
