SseIm Framework

im:
  endpoints:
    - path: /api/chat/pull   # URL 路径（对外契约）
      topic: im.chat           # 内部 Topic（业务语义）
      mode: pull               # PUSH 或 PULL

im:
  endpoints:
    - path: /api/chat/push
      topic: im.chat
      mode: push              # PUSH: POST → MQ
    - path: /api/chat/pull
      topic: im.chat
      mode: pull              # PULL: GET → SSE

// 1. 继承骨架类
public class ChatPublisher extends ImPublisher {
    public ChatPublisher(String topic, Supplier<Repeater> repeaterSupplier) {
        super(topic, repeaterSupplier);
    }

    @Override
    public Serializable push(MultiValueMap<String, String> urlParams, Msg msg) {
        log.info("[ChatPublisher] 推送聊天消息");
        return super.push(urlParams, msg);
    }
}

// 2. 实现工厂
@Component
public class ChatPublisherFactory implements PublisherFactory {
    @Override
    public Publisher create(String topic, Supplier<Repeater> repeaterSupplier) {
        if ("im.chat".equals(topic)) {
            return new ChatPublisher(topic, repeaterSupplier);
        }
        return null; // 其他 Topic 使用默认实现
    }
}

@Component
public class RedisReplayInterceptor implements RepeaterInterceptor {
    @Autowired
    private RedisTemplate<String, byte> redisTemplate;

    @Override
    public Mono<Boolean> beforeSend(TopicMessage<Msg> message) {
        // 发送前持久化到 Redis
        redisTemplate.opsForList().rightPush(key(message.getTopic()), serialize(message));
        return Mono.just(true); // false 拦截消息
    }

    @Override
    public Flux<Msg> transformSubscribeStream(String topic, Flux<Msg> stream, ...) {
        // 订阅时先回放历史消息
        Flux<Msg> history = loadFromRedis(topic);
        return Flux.concat(history, stream);
    }
}

@Component
@Order(10)
public class JwtAuthFilter implements AccessFilter {
    @Autowired
    private JwtService jwtService;

    @Override
    public void doFilter(AccessContext context, AccessFilterChain chain) {
        String token = context.getHeader("Authorization")
                .filter(h -> h.startsWith("Bearer "))
                .map(h -> h.substring(7))
                .orElseThrow(() -> new AccessDeniedException("Missing token", HttpStatus.UNAUTHORIZED));

        UserDetails user = jwtService.validate(token);
        context.setPrincipal(user);
        chain.doFilter(context);
    }
}

@RestController
public class AdminController {
    @Autowired
    private EndpointRegistry endpointRegistry;

    @PostMapping("/admin/endpoint")
    public R<?> addEndpoint(@RequestBody EndpointProperties config) {
        endpointRegistry.register(config);  // 即时生效
        return R.ok();
    }
}

@Data
public class ChatMsg implements Msg {
    private String from;
    private String content;
    private long timestamp;
    // 无需实现 build()，框架提供默认实现
}

public class DirectRepeater implements Repeater {
    @Override
    public Serializable send(TopicMessage<Msg> message) {
        // 直接推送到 Redis Pub/Sub，绕过 MQ
        redisTemplate.convertAndSend(message.getTopic(), serialize(message));
        return message.getId();
    }

    @Override
    public void receive(TopicMessage<Msg> message) {
        // 从 Redis 接收后推送到 SSE
        sseConnectionManager.publish(message.getTopic(), message);
    }

    @Override
    public Flux<Msg> subscribe(MultiValueMap<String, String> urlParams) {
        // 从 Redis 订阅实时流
        return redisReactiveTemplate.listenToChannel(topic()).map(this::deserialize);
    }
}

// 单键定向 — 投递到 lobby 房间
// ?room=lobby → targets = {"room=lobby"}
fetch('/api/chat/push?room=lobby', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({from: 'bob', content: 'hi lobby'})
});

// 单键多值 — 批量投递到多个房间
// ?room=lobby&room=main → targets = {"room=lobby", "room=main"}
fetch('/api/chat/push?room=lobby&room=main', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({from: 'charlie', content: '群聊'})
});

// 多键位置配对 — 多租户场景（需端点配置 routing-keys: ["tenant", "room"]）
fetch('/api/room/push?tenant=acme&room=lobby', {
  method: 'POST',
  headers: {'Content-Type': 'application/json'},
  body: JSON.stringify({from: 'system', content: '房间通知'})
});

// 无参数 = 广播
fetch('/api/chat/push', { method: 'POST', ... });

@Autowired
private TopicRegistry topicRegistry;

@Autowired
private RoutingKeyComposer composer;

// 定向投递 — Composer 构建 routingKey，与 HTTP 路径格式一致
public void sendToRoom(String room, ChatMsg msg) {
    MultiValueMap<String, String> params = new LinkedMultiValueMap<>();
    params.add("room", room);
    Set<String> targets = composer.extractRoutingTargets(null, params);

    TopicMessage<ChatMsg> envelope = TopicMessage.<ChatMsg>builder()
            .topic("im.chat")
            .msg(msg)
            .routingKeys(targets)
            .build();
    topicRegistry.ensureRepeater("im.chat").send(envelope);
}

// 批量投递
public void sendToRooms(Set<String> rooms, ChatMsg msg) {
    MultiValueMap<String, String> params = new LinkedMultiValueMap<>();
    rooms.forEach(room -> params.add("room", room));
    Set<String> targets = composer.extractRoutingTargets(null, params);

    TopicMessage<ChatMsg> envelope = TopicMessage.<ChatMsg>builder()
            .topic("im.chat")
            .msg(msg)
            .routingKeys(targets)
            .build();
    topicRegistry.ensureRepeater("im.chat").send(envelope);
}

im:
  sse:
    routing-keys: [room]       # 全局默认（按房间路由）
  endpoints:
    - path: /api/chat/pull
      topic: im.chat
      mode: pull               # 继承全局 ["room"]
    - path: /api/chat/push
      topic: im.chat
      mode: push               # 继承全局 ["room"]
    - path: /api/room/pull
      topic: im.room
      mode: pull
      routing-keys: [tenant, room]  # 端点覆盖：多租户聊天室
    - path: /api/room/push
      topic: im.room
      mode: push
      routing-keys: [tenant, room]  # 必须与对应 pull 端点一致

@Component
public class SignedRoutingKeyComposer implements RoutingKeyComposer {

    @Override
    public String composeRoutingKey(List<String> routingKeys,
            MultiValueMap<String, String> params) {
        String raw = buildRawKey(routingKeys, params);
        return raw + "&sig=" + hmacSign(raw);
    }

    @Override
    public Set<String> extractRoutingTargets(List<String> routingKeys,
            MultiValueMap<String, String> params) {
        // 需与 composeRoutingKey 保持寻址语义一致
    }
}

@Component
public class UserDetailsRoutingKeyConverter
        implements PrincipalRoutingKeyConverter {

    @Override
    public boolean supports(Class<?> principalType) {
        return UserDetails.class.isAssignableFrom(principalType);
    }

    @Override
    public String convert(Object principal, AccessContext context) {
        return ((UserDetails) principal).getUsername();
    }
}

@Component
@Order(10)
public class JwtAuthFilter implements AccessFilter {

    @Autowired
    private JwtService jwtService;

    @Override
    public void doFilter(AccessContext context, AccessFilterChain chain) {
        String token = context.getHeader("Authorization")
                .filter(h -> h.startsWith("Bearer "))
                .map(h -> h.substring(7))
                .orElseThrow(() -> new AccessDeniedException(
                        "Missing token", HttpStatus.UNAUTHORIZED));

        String userId = jwtService.parseUserId(token);
        context.setPrincipal(userId);
        chain.doFilter(context);
    }
}

import { fetchEventSource } from '@microsoft/fetch-event-source';

await fetchEventSource('/api/chat/pull', {
  headers: { 'Authorization': 'Bearer eyJhbG...' },
  onmessage(ev) { console.log(JSON.parse(ev.data)); }
});

logging:
  level:
    cn.gmlee.tools.im: DEBUG      # 框架核心日志
    cn.gmlee.tools.im.sse: DEBUG  # SSE 连接管理日志