package com.ard.gb28181.transmit.event; import com.ard.gb28181.api.bean.DeviceNotFoundEvent; import com.ard.gb28181.transmit.event.sip.SipEvent; import gov.nist.javax.sip.message.SIPRequest; import gov.nist.javax.sip.message.SIPResponse; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.ObjectUtils; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import javax.sip.DialogTerminatedEvent; import javax.sip.ResponseEvent; import javax.sip.TimeoutEvent; import javax.sip.TransactionTerminatedEvent; import javax.sip.header.WarningHeader; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.DelayQueue; /** * @author lin */ @Slf4j @Component public class SipSubscribe { private final Map subscribes = new ConcurrentHashMap<>(); private final DelayQueue delayQueue = new DelayQueue<>(); @Scheduled(fixedDelay = 200) public void execute() { // 1. 使用 poll() 替代 take() // poll() 会立即返回:如果有已超期的任务则返回,否则返回 null SipEvent take = delayQueue.poll(); // 2. 循环处理所有已超期的任务 while (take != null) { try { // 出现超时异常 if (take.getErrorEvent() != null) { EventResult eventResult = new EventResult<>(); eventResult.type = EventResultType.timeout; eventResult.msg = "消息超时未回复"; eventResult.statusCode = -1024; take.getErrorEvent().response(eventResult); } subscribes.remove(take.getKey()); } catch (Exception e) { // 建议捕获 Exception,防止单个任务异常导致循环中断 log.error("[SIP超时检查] 处理超时任务异常", e); } // 3. 继续尝试获取下一个已超期的任务 take = delayQueue.poll(); } // 4. 如果 poll() 返回 null,说明当前没有超时的任务,方法直接结束 // 线程释放,等待下一次 200ms 调度 } public interface Event { void response(EventResult eventResult); } /** * */ public enum EventResultType { // 超时 timeout, // 回复 response, // 事务已结束 transactionTerminated, // 会话已结束 dialogTerminated, // 设备未找到 deviceNotFoundEvent, // 消息发送失败 cmdSendFailEvent, // 消息发送失败 failedToGetPort, // 收到失败的回复 failedResult } public static class EventResult { public int statusCode; public EventResultType type; public String msg; public String callId; public T event; public EventResult() { } public EventResult(T event) { this.event = event; if (event instanceof ResponseEvent) { ResponseEvent responseEvent = (ResponseEvent) event; SIPResponse response = (SIPResponse) responseEvent.getResponse(); this.type = EventResultType.response; if (response != null) { WarningHeader warningHeader = (WarningHeader) response.getHeader(WarningHeader.NAME); if (warningHeader != null && !ObjectUtils.isEmpty(warningHeader.getText())) { this.msg = ""; if (warningHeader.getCode() > 0) { this.msg += warningHeader.getCode() + ":"; } if (warningHeader.getAgent() != null) { this.msg += warningHeader.getCode() + ":"; } if (warningHeader.getText() != null) { this.msg += warningHeader.getText(); } } else { this.msg = response.getReasonPhrase(); } this.statusCode = response.getStatusCode(); this.callId = response.getCallIdHeader().getCallId(); } } else if (event instanceof TimeoutEvent) { TimeoutEvent timeoutEvent = (TimeoutEvent) event; this.type = EventResultType.timeout; this.msg = "消息超时未回复"; this.statusCode = -1024; if (timeoutEvent.isServerTransaction()) { this.callId = ((SIPRequest) timeoutEvent.getServerTransaction().getRequest()).getCallIdHeader().getCallId(); } else { this.callId = ((SIPRequest) timeoutEvent.getClientTransaction().getRequest()).getCallIdHeader().getCallId(); } } else if (event instanceof TransactionTerminatedEvent) { TransactionTerminatedEvent transactionTerminatedEvent = (TransactionTerminatedEvent) event; this.type = EventResultType.transactionTerminated; this.msg = "事务已结束"; this.statusCode = -1024; if (transactionTerminatedEvent.isServerTransaction()) { this.callId = ((SIPRequest) transactionTerminatedEvent.getServerTransaction().getRequest()).getCallIdHeader().getCallId(); } else { this.callId = ((SIPRequest) transactionTerminatedEvent.getClientTransaction().getRequest()).getCallIdHeader().getCallId(); } } else if (event instanceof DialogTerminatedEvent) { DialogTerminatedEvent dialogTerminatedEvent = (DialogTerminatedEvent) event; this.type = EventResultType.dialogTerminated; this.msg = "会话已结束"; this.statusCode = -1024; this.callId = dialogTerminatedEvent.getDialog().getCallId().getCallId(); } else if (event instanceof DeviceNotFoundEvent) { this.type = EventResultType.deviceNotFoundEvent; this.msg = "设备未找到"; this.statusCode = -1024; this.callId = ((DeviceNotFoundEvent) event).getCallId(); } } } public void addSubscribe(String key, SipEvent event) { SipEvent sipEvent = subscribes.get(key); if (sipEvent != null) { subscribes.remove(key); delayQueue.remove(sipEvent); } subscribes.put(key, event); delayQueue.offer(event); } public SipEvent getSubscribe(String key) { return subscribes.get(key); } public void removeSubscribe(String key) { if (key == null) { return; } SipEvent sipEvent = subscribes.get(key); if (sipEvent != null) { subscribes.remove(key); delayQueue.remove(sipEvent); } } public boolean isEmpty() { return subscribes.isEmpty(); } public Integer size() { return subscribes.size(); } }