|
|
|
@ -3,7 +3,7 @@ package com.cloud.kicc.commonbiz.service.impl; |
|
|
|
import cn.hutool.core.collection.ConcurrentHashSet; |
|
|
|
import cn.hutool.core.collection.ConcurrentHashSet; |
|
|
|
import cn.hutool.core.util.StrUtil; |
|
|
|
import cn.hutool.core.util.StrUtil; |
|
|
|
import com.cloud.kicc.common.core.exception.CheckedException; |
|
|
|
import com.cloud.kicc.common.core.exception.CheckedException; |
|
|
|
import com.cloud.kicc.common.data.entity.CasUser; |
|
|
|
import com.cloud.kicc.common.data.entity.KiccUser; |
|
|
|
import com.cloud.kicc.common.security.util.SecurityUtils; |
|
|
|
import com.cloud.kicc.common.security.util.SecurityUtils; |
|
|
|
import com.cloud.kicc.commonbiz.api.entity.SseSignalContainer; |
|
|
|
import com.cloud.kicc.commonbiz.api.entity.SseSignalContainer; |
|
|
|
import com.cloud.kicc.commonbiz.service.IMapLogisticSseService; |
|
|
|
import com.cloud.kicc.commonbiz.service.IMapLogisticSseService; |
|
|
|
@ -48,57 +48,57 @@ public class MapLogisticSseServiceImpl implements IMapLogisticSseService { |
|
|
|
@Override |
|
|
|
@Override |
|
|
|
@SneakyThrows |
|
|
|
@SneakyThrows |
|
|
|
public SseEmitter SseSubscribe(String clientId) { |
|
|
|
public SseEmitter SseSubscribe(String clientId) { |
|
|
|
CasUser casUser = getUser(); |
|
|
|
KiccUser user = getUser(); |
|
|
|
Optional.ofNullable(clientId).orElseThrow(() -> new CheckedException("当前客户端Id为空,请检查后重试!")); |
|
|
|
Optional.ofNullable(clientId).orElseThrow(() -> new CheckedException("当前客户端Id为空,请检查后重试!")); |
|
|
|
// 设置超时时间为1小时
|
|
|
|
// 设置超时时间为1小时
|
|
|
|
SseEmitter sseEmitter = new SseEmitter(3600_000L); |
|
|
|
SseEmitter sseEmitter = new SseEmitter(3600_000L); |
|
|
|
SseSignalContainer sseSignalContainer =new SseSignalContainer( |
|
|
|
SseSignalContainer sseSignalContainer =new SseSignalContainer( |
|
|
|
clientId, |
|
|
|
clientId, |
|
|
|
casUser.getId(), |
|
|
|
user.getId(), |
|
|
|
sseEmitter, |
|
|
|
sseEmitter, |
|
|
|
casUser.getTenantId() |
|
|
|
user.getTenantId() |
|
|
|
); |
|
|
|
); |
|
|
|
// 设置如果网络出错前端请求的重试时间为1s
|
|
|
|
// 设置如果网络出错前端请求的重试时间为1s
|
|
|
|
sseEmitter.send(SseEmitter.event().data("创建通道连接成功").reconnectTime(1000)); |
|
|
|
sseEmitter.send(SseEmitter.event().data("创建通道连接成功").reconnectTime(1000)); |
|
|
|
sseSignalContainers.add(sseSignalContainer); |
|
|
|
sseSignalContainers.add(sseSignalContainer); |
|
|
|
log.info("clientId:{},建立的用户Id为:{}", clientId, casUser.getId()); |
|
|
|
log.info("clientId:{},建立的用户Id为:{}", clientId, user.getId()); |
|
|
|
sseEmitter.onTimeout(() -> { |
|
|
|
sseEmitter.onTimeout(() -> { |
|
|
|
log.info("clientId:{},用户Id为:{},的SSE长轮询已经超时,正在删除当前的建立通道对象", clientId, casUser.getId()); |
|
|
|
log.info("clientId:{},用户Id为:{},的SSE长轮询已经超时,正在删除当前的建立通道对象", clientId, user.getId()); |
|
|
|
sseEmitter.complete(); |
|
|
|
sseEmitter.complete(); |
|
|
|
sseSignalContainers.remove(sseSignalContainer); |
|
|
|
sseSignalContainers.remove(sseSignalContainer); |
|
|
|
}); |
|
|
|
}); |
|
|
|
sseEmitter.onCompletion(() -> { |
|
|
|
sseEmitter.onCompletion(() -> { |
|
|
|
log.info("clientId:{},用户Id为:{}的SSE长轮询已经返回响应关闭,正在删除当前的建立通道对象", clientId, casUser.getId()); |
|
|
|
log.info("clientId:{},用户Id为:{}的SSE长轮询已经返回响应关闭,正在删除当前的建立通道对象", clientId, user.getId()); |
|
|
|
sseSignalContainers.remove(sseSignalContainer); |
|
|
|
sseSignalContainers.remove(sseSignalContainer); |
|
|
|
}); |
|
|
|
}); |
|
|
|
sseEmitter.onError(e -> log.info("clientId:{},当前用户Id为:{}的SSE长轮询出现异常,正在删除当前的建立通道对象,错误信息{}", clientId, casUser.getId(), e.getLocalizedMessage())); |
|
|
|
sseEmitter.onError(e -> log.info("clientId:{},当前用户Id为:{}的SSE长轮询出现异常,正在删除当前的建立通道对象,错误信息{}", clientId, user.getId(), e.getLocalizedMessage())); |
|
|
|
return sseEmitter; |
|
|
|
return sseEmitter; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
@Override |
|
|
|
public void sendMessage(String userId, SseEmitterConsumer<SseSignalContainer> consumer) { |
|
|
|
public void sendMessage(String userId, SseEmitterConsumer<SseSignalContainer> consumer) { |
|
|
|
CasUser casUser = getUser(); |
|
|
|
KiccUser user = getUser(); |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
.filter(item -> StrUtil.equals(item.getUserId(), userId) && StrUtil.equals(item.getTenantId(), casUser.getTenantId())) |
|
|
|
.filter(item -> StrUtil.equals(item.getUserId(), userId) && StrUtil.equals(item.getTenantId(), user.getTenantId())) |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
send(sendSseSignalContainers, consumer); |
|
|
|
send(sendSseSignalContainers, consumer); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
@Override |
|
|
|
public void sendMessage(SseEmitterConsumer<SseSignalContainer> consumer) { |
|
|
|
public void sendMessage(SseEmitterConsumer<SseSignalContainer> consumer) { |
|
|
|
CasUser casUser = getUser(); |
|
|
|
KiccUser user = getUser(); |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
.filter(item -> StrUtil.equals(item.getTenantId(), casUser.getTenantId())) |
|
|
|
.filter(item -> StrUtil.equals(item.getTenantId(), user.getTenantId())) |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
send(sendSseSignalContainers, consumer); |
|
|
|
send(sendSseSignalContainers, consumer); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
@Override |
|
|
|
public void disconnect(String clientId) { |
|
|
|
public void disconnect(String clientId) { |
|
|
|
CasUser casUser = getUser(); |
|
|
|
KiccUser user = getUser(); |
|
|
|
Optional.ofNullable(clientId).orElseThrow(() -> new CheckedException("当前客户端Id为空,请检查后重试!")); |
|
|
|
Optional.ofNullable(clientId).orElseThrow(() -> new CheckedException("当前客户端Id为空,请检查后重试!")); |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
.filter(item -> StrUtil.equals(item.getClientId(), clientId) && StrUtil.equals(item.getTenantId(), casUser.getTenantId())) |
|
|
|
.filter(item -> StrUtil.equals(item.getClientId(), clientId) && StrUtil.equals(item.getTenantId(), user.getTenantId())) |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
sendSseSignalContainers.forEach(item -> item.getSseEmitter().complete()); |
|
|
|
sendSseSignalContainers.forEach(item -> item.getSseEmitter().complete()); |
|
|
|
sseSignalContainers.removeAll(sendSseSignalContainers); |
|
|
|
sseSignalContainers.removeAll(sendSseSignalContainers); |
|
|
|
@ -106,9 +106,9 @@ public class MapLogisticSseServiceImpl implements IMapLogisticSseService { |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
|
@Override |
|
|
|
public void disconnect() { |
|
|
|
public void disconnect() { |
|
|
|
CasUser casUser = getUser(); |
|
|
|
KiccUser user = getUser(); |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
Set<SseSignalContainer> sendSseSignalContainers = sseSignalContainers.stream() |
|
|
|
.filter(item -> StrUtil.equals(item.getTenantId(), casUser.getTenantId())) |
|
|
|
.filter(item -> StrUtil.equals(item.getTenantId(), user.getTenantId())) |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
.collect(Collectors.toSet()); |
|
|
|
sendSseSignalContainers.forEach(item -> item.getSseEmitter().complete()); |
|
|
|
sendSseSignalContainers.forEach(item -> item.getSseEmitter().complete()); |
|
|
|
sseSignalContainers.removeAll(sendSseSignalContainers); |
|
|
|
sseSignalContainers.removeAll(sendSseSignalContainers); |
|
|
|
@ -154,10 +154,10 @@ public class MapLogisticSseServiceImpl implements IMapLogisticSseService { |
|
|
|
return failedEmitters; |
|
|
|
return failedEmitters; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private CasUser getUser() { |
|
|
|
private KiccUser getUser() { |
|
|
|
CasUser casUser = SecurityUtils.getCasUser(); |
|
|
|
KiccUser user = SecurityUtils.getUser(); |
|
|
|
Optional.ofNullable(casUser).orElseThrow(() -> new CheckedException("当前用户登录,请先登录后重试!")); |
|
|
|
Optional.ofNullable(user).orElseThrow(() -> new CheckedException("当前用户登录,请先登录后重试!")); |
|
|
|
return casUser; |
|
|
|
return user; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
|
wangxiang
2 years ago