package com.liquidnet.common.cache.redis.config;

import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.time.Duration;
import java.util.HashMap;

@Slf4j
public class RedisStreamConfig {
    private String hostname;
    {
        try {
            hostname = InetAddress.getLocalHost().getHostName() + "_";
        } catch (UnknownHostException ignored) {
            hostname = "";
        }
    }

    public void initStream(StringRedisTemplate stringRedisTemplate, String key, String group) {
        try {
            if (stringRedisTemplate.hasKey(key)) {
                log.info("redis stream exist[{},{}]", key, group);
                stringRedisTemplate.opsForStream().createGroup(key, group);
            } else {
                log.info("redis stream init[{},{}]", key, group);
                HashMap<String, String> map = new HashMap<>();
                map.put("message", " ");
                StreamOperations<String, Object, Object> stringObjectObjectStreamOperations = stringRedisTemplate.opsForStream();
                RecordId recordId = stringObjectObjectStreamOperations.add(StreamRecords.mapBacked(map).withStreamKey(key));
                stringObjectObjectStreamOperations.createGroup(key, group);
                stringObjectObjectStreamOperations.delete(key, recordId.getValue());
            }
        } catch (Exception e) {
            log.warn("###Ex:redis stream init [{},{}],ex:{}", key, group, e.getMessage());
        }
    }

    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> buildStreamMessageListenerContainer(RedisConnectionFactory factory) {
//        var options = StreamMessageListenerContainer
//                .StreamMessageListenerContainerOptions
//                .builder()
//                .pollTimeout(Duration.ofMillis(1))
//                .build();

        // 创建配置对象
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次性最多拉取多少条消息
                .batchSize(1)
                // 执行消息轮询的执行器
//                .executor(this.threadPoolTaskExecutor)
                // 消息消费异常的handler
                .errorHandler(t -> {
                    // StreamListener中的异常会在这里抛出
                    System.out.println("errorHandler: " + t.getMessage());
                })
                // 超时时间，设置为0，表示不超时（超时后会抛出异常）
                .pollTimeout(Duration.ofMillis(1))
                .build();
        return StreamMessageListenerContainer.create(factory, options);
    }

    public String getConsumerName(String consumerSeqname) {
        return hostname.concat(consumerSeqname);
    }
}
