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

import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
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 org.springframework.stereotype.Component;

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.error("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();
        return StreamMessageListenerContainer.create(factory, options);
    }

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