@RequiredArgsConstructor
@Service
public class StreamProducer {

    private final RedisTemplate<String, Object> redisTemplate;

    // Map 기반 레코드 추가
    public RecordId send(String streamKey, Map<String, String> data) {
        StringRecord record = StreamRecords.string(data)
            .withStreamKey(streamKey);
        return redisTemplate.opsForStream().add(record);
    }

    // Object 기반 레코드 추가
    public RecordId send(String streamKey, Object payload) {
        ObjectRecord<String, Object> record = StreamRecords.newRecord()
            .in(streamKey)
            .ofObject(payload);
        return redisTemplate.opsForStream().add(record);
    }
}

StringRecord record = StreamRecords.string(data).withStreamKey("my-stream");
redisTemplate.opsForStream().add(record, StreamAddOptions.maxLen(1000).approximate());

// 최신 메시지 읽기
List<MapRecord<String, String, String>> messages = redisTemplate.opsForStream()
    .read(StreamReadOptions.empty().count(10),
          StreamOffset.latest("my-stream"));

// Consumer Group에서 읽기
List<MapRecord<String, String, String>> messages = redisTemplate.opsForStream()
    .read(Consumer.from("my-group", "my-consumer"),
          StreamReadOptions.empty().count(10),
          StreamOffset.create("my-stream", ReadOffset.lastConsumed()));

@Slf4j
@Component
public class OrderEventListener implements StreamListener<String, MapRecord<String, String, String>> {

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        log.info("Message ID: {}", message.getId());
        log.info("Stream: {}", message.getStream());
        log.info("Body: {}", message.getValue());
    }
}

@Configuration
public class RedisStreamConfig {

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer(
            RedisConnectionFactory connectionFactory,
            OrderEventListener listener) {

        var options = StreamMessageListenerContainerOptions.builder()
            .pollTimeout(Duration.ofMillis(100))
            .batchSize(10)
            .build();

        var container = StreamMessageListenerContainer.create(connectionFactory, options);

        // 단순 구독 (Consumer Group 없이)
        container.receive(
            StreamOffset.fromStart("my-stream"),
            listener
        );

        // Consumer Group으로 구독
        container.receive(
            Consumer.from("my-group", "my-consumer"),
            StreamOffset.create("my-stream", ReadOffset.lastConsumed()),
            listener
        );

        // 자동 ACK로 구독
        container.receiveAutoAck(
            Consumer.from("my-group", "my-consumer"),
            StreamOffset.create("my-stream", ReadOffset.lastConsumed()),
            listener
        );

        container.start();
        return container;
    }
}

@RequiredArgsConstructor
@Service
public class StreamGroupManager {

    private final RedisTemplate<String, Object> redisTemplate;

    // Consumer Group 생성
    public void createGroup(String streamKey, String groupName) {
        try {
            redisTemplate.opsForStream()
                .createGroup(streamKey, ReadOffset.from("0"), groupName);
        } catch (RedisSystemException e) {
            // BUSYGROUP: 이미 존재하는 그룹
            log.warn("Consumer group already exists: {}", groupName);
        }
    }

    // 메시지 ACK
    public void acknowledge(String streamKey, String groupName, RecordId... recordIds) {
        redisTemplate.opsForStream().acknowledge(streamKey, groupName, recordIds);
    }

    // Pending 메시지 조회
    public PendingMessagesSummary pending(String streamKey, String groupName) {
        return redisTemplate.opsForStream().pending(streamKey, groupName);
    }
}

// 도메인 객체
@Data
public class OrderEvent {
    private String orderId;
    private String type;
    private String userId;
}

// ObjectRecord로 추가
ObjectRecord<String, OrderEvent> record = StreamRecords.newRecord()
    .in("order-events")
    .ofObject(new OrderEvent("12345", "CREATED", "user1"));
redisTemplate.opsForStream().add(record);

// ObjectRecord로 읽기
List<ObjectRecord<String, OrderEvent>> records = redisTemplate.opsForStream()
    .read(OrderEvent.class, StreamOffset.fromStart("order-events"));

@Component
public class OrderEventObjectListener
        implements StreamListener<String, ObjectRecord<String, OrderEvent>> {

    @Override
    public void onMessage(ObjectRecord<String, OrderEvent> message) {
        OrderEvent event = message.getValue();
        // event.getOrderId(), event.getType() ...
    }
}

// Container에 ObjectMapper 설정
var options = StreamMessageListenerContainerOptions.builder()
    .pollTimeout(Duration.ofMillis(100))
    .targetType(OrderEvent.class)  // 자동 매핑
    .build();

ReactiveRedisConnectionFactory connectionFactory = ...;

var options = StreamReceiverOptions.builder()
    .pollTimeout(Duration.ofMillis(100))
    .build();

StreamReceiver<String, MapRecord<String, String, String>> receiver =
    StreamReceiver.create(connectionFactory, options);

Flux<MapRecord<String, String, String>> messages =
    receiver.receive(StreamOffset.fromStart("my-stream"));

messages
    .doOnNext(message -> {
        log.info("ID: {}, Body: {}", message.getId(), message.getValue());
    })
    .subscribe();