package com.liquidnet.service.dragon.config;

import com.liquidnet.service.dragon.receiver.RedisPayReceiver;
import com.liquidnet.service.dragon.receiver.RedisRefundReceiver;
import lombok.var;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import java.time.Duration;

@Configuration
public class RedisStreamConfig {

    @Autowired
    private RedisPayReceiver redisPayReceiver;
    @Autowired
    private RedisRefundReceiver redisRefundReceiver;

    @Bean
    public Subscription subscriptionPay0(RedisConnectionFactory factory) {
        var options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofMillis(1))
                .build();
        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
        var subscription = listenerContainer.receiveAutoAck(Consumer.from("dragon-pay-group", "dragon-pay-0"),
                StreamOffset.create("dragon-pay", ReadOffset.lastConsumed()), redisPayReceiver);
        listenerContainer.start();
        return subscription;
    }

    @Bean
    public Subscription subscriptionPay1(RedisConnectionFactory factory) {
        var options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofMillis(1))
                .build();
        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
        var subscription = listenerContainer.receiveAutoAck(Consumer.from("dragon-pay-group", "dragon-pay-1"),
                StreamOffset.create("dragon-pay", ReadOffset.lastConsumed()), redisPayReceiver);
        listenerContainer.start();
        return subscription;
    }

    @Bean
    public Subscription subscriptionRefund0(RedisConnectionFactory factory) {
        var options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofMillis(1))
                .build();
        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
        var subscription = listenerContainer.receiveAutoAck(Consumer.from("dragon-refund-group", "dragon-refund-0"),
                StreamOffset.create("dragon-refund", ReadOffset.lastConsumed()), redisRefundReceiver);
        listenerContainer.start();
        return subscription;
    }

    @Bean
    public Subscription subscriptionRefund1(RedisConnectionFactory factory) {
        var options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofMillis(1))
                .build();
        var listenerContainer = StreamMessageListenerContainer.create(factory, options);
        var subscription = listenerContainer.receiveAutoAck(Consumer.from("dragon-refund-group", "dragon-refund-1"),
                StreamOffset.create("dragon-refund", ReadOffset.lastConsumed()), redisRefundReceiver);
        listenerContainer.start();
        return subscription;
    }
}
