From 28b5cc39d0a2d9939f70b4c980a31d9b27fc1e4c Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 19 七月 2022 14:27:30 +0800
Subject: [PATCH] 修复更新推流状态sql错误
---
src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java | 88 ++++++++++++++++++++++++++++++++-----------
1 files changed, 65 insertions(+), 23 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java
index 48eebf8..a2fab81 100644
--- a/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java
+++ b/src/main/java/com/genersoft/iot/vmp/gb28181/transmit/SIPProcessorObserver.java
@@ -1,19 +1,26 @@
package com.genersoft.iot.vmp.gb28181.transmit;
+import com.genersoft.iot.vmp.gb28181.event.EventPublisher;
import com.genersoft.iot.vmp.gb28181.event.SipSubscribe;
import com.genersoft.iot.vmp.gb28181.transmit.event.request.ISIPRequestProcessor;
+import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.RegisterRequestProcessor;
+import com.genersoft.iot.vmp.gb28181.transmit.event.request.impl.message.notify.cmd.KeepaliveNotifyMessageHandler;
import com.genersoft.iot.vmp.gb28181.transmit.event.response.ISIPResponseProcessor;
import com.genersoft.iot.vmp.gb28181.transmit.event.timeout.ITimeoutProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import javax.sip.*;
-import javax.sip.header.CSeqHeader;
-import javax.sip.header.CallIdHeader;
+import javax.sip.address.SipURI;
+import javax.sip.address.URI;
+import javax.sip.header.*;
+import javax.sip.message.Request;
import javax.sip.message.Response;
import java.util.Map;
+import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
/**
@@ -22,7 +29,7 @@
* @date: 2021骞�11鏈�5鏃� 涓嬪崍15锛�32
*/
@Component
-public class SIPProcessorObserver implements SipListener {
+public class SIPProcessorObserver implements ISIPProcessorObserver {
private final static Logger logger = LoggerFactory.getLogger(SIPProcessorObserver.class);
@@ -32,6 +39,12 @@
@Autowired
private SipSubscribe sipSubscribe;
+
+ @Autowired
+ private EventPublisher eventPublisher;
+
+
+
/**
* 娣诲姞 request璁㈤槄
@@ -56,7 +69,7 @@
* @param processor 澶勭悊绋嬪簭
*/
public void addTimeoutProcessor(ITimeoutProcessor processor) {
- this.timeoutProcessor = processor;
+ timeoutProcessor = processor;
}
/**
@@ -64,7 +77,9 @@
* @param requestEvent RequestEvent浜嬩欢
*/
@Override
+ @Async
public void processRequest(RequestEvent requestEvent) {
+ logger.debug("\n鏀跺埌璇锋眰锛歕n{}", requestEvent.getRequest());
String method = requestEvent.getRequest().getMethod();
ISIPRequestProcessor sipRequestProcessor = requestProcessorMap.get(method);
if (sipRequestProcessor == null) {
@@ -72,6 +87,7 @@
return;
}
requestProcessorMap.get(method).process(requestEvent);
+
}
/**
@@ -79,35 +95,26 @@
* @param responseEvent responseEvent浜嬩欢
*/
@Override
+ @Async
public void processResponse(ResponseEvent responseEvent) {
- logger.debug(responseEvent.getResponse().toString());
-// CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
-// String method = cseqHeader.getMethod();
-// ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
-// if (sipRequestProcessor == null) {
-// logger.warn("涓嶆敮鎸佹柟娉晎}鐨剅esponse", method);
-// return;
-// }
-// sipRequestProcessor.process(responseEvent);
-
-
Response response = responseEvent.getResponse();
- logger.debug(responseEvent.getResponse().toString());
+ logger.debug("\n鏀跺埌鍝嶅簲锛歕n{}", responseEvent.getResponse());
int status = response.getStatusCode();
- if (((status >= 200) && (status < 300)) || status == 401) { // Success!
-// ISIPResponseProcessor processor = processorFactory.createResponseProcessor(evt);
+
+ if (((status >= 200) && (status < 300)) || status == Response.UNAUTHORIZED) { // Success!
CSeqHeader cseqHeader = (CSeqHeader) responseEvent.getResponse().getHeader(CSeqHeader.NAME);
String method = cseqHeader.getMethod();
ISIPResponseProcessor sipRequestProcessor = responseProcessorMap.get(method);
if (sipRequestProcessor != null) {
sipRequestProcessor.process(responseEvent);
}
- if (responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
+ if (status != Response.UNAUTHORIZED && responseEvent.getResponse() != null && sipSubscribe.getOkSubscribesSize() > 0 ) {
CallIdHeader callIdHeader = (CallIdHeader)responseEvent.getResponse().getHeader(CallIdHeader.NAME);
if (callIdHeader != null) {
SipSubscribe.Event subscribe = sipSubscribe.getOkSubscribe(callIdHeader.getCallId());
if (subscribe != null) {
SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
+ sipSubscribe.removeOkSubscribe(callIdHeader.getCallId());
subscribe.response(eventResult);
}
}
@@ -123,10 +130,15 @@
if (subscribe != null) {
SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(responseEvent);
subscribe.response(eventResult);
+ sipSubscribe.removeErrorSubscribe(callIdHeader.getCallId());
}
}
}
+ if (responseEvent.getDialog() != null) {
+ responseEvent.getDialog().delete();
+ }
}
+
}
@@ -136,25 +148,55 @@
*/
@Override
public void processTimeout(TimeoutEvent timeoutEvent) {
- if(timeoutProcessor != null) {
- timeoutProcessor.process(timeoutEvent);
+ logger.info("[娑堟伅鍙戦�佽秴鏃禲");
+ ClientTransaction clientTransaction = timeoutEvent.getClientTransaction();
+
+ if (clientTransaction != null) {
+ logger.info("[鍙戦�侀敊璇闃匽 clientTransaction != null");
+ Request request = clientTransaction.getRequest();
+ if (request != null) {
+ logger.info("[鍙戦�侀敊璇闃匽 request != null");
+ CallIdHeader callIdHeader = (CallIdHeader) request.getHeader(CallIdHeader.NAME);
+ if (callIdHeader != null) {
+ logger.info("[鍙戦�侀敊璇闃匽");
+ SipSubscribe.Event subscribe = sipSubscribe.getErrorSubscribe(callIdHeader.getCallId());
+ SipSubscribe.EventResult eventResult = new SipSubscribe.EventResult(timeoutEvent);
+ subscribe.response(eventResult);
+ sipSubscribe.removeOkSubscribe(callIdHeader.getCallId());
+ sipSubscribe.removeErrorSubscribe(callIdHeader.getCallId());
+ }
+ }
}
+ eventPublisher.requestTimeOut(timeoutEvent);
}
@Override
public void processIOException(IOExceptionEvent exceptionEvent) {
-// System.out.println("processIOException");
+ System.out.println("processIOException");
}
@Override
public void processTransactionTerminated(TransactionTerminatedEvent transactionTerminatedEvent) {
+// Transaction transaction = null;
// System.out.println("processTransactionTerminated");
+// if (transactionTerminatedEvent.isServerTransaction()) {
+// transaction = transactionTerminatedEvent.getServerTransaction();
+// }else {
+// transaction = transactionTerminatedEvent.getClientTransaction();
+// }
+//
+// System.out.println(transaction.getBranchId());
+// System.out.println(transaction.getState());
+// System.out.println(transaction.getRequest().getMethod());
+// CallIdHeader header = (CallIdHeader)transaction.getRequest().getHeader(CallIdHeader.NAME);
+// SipSubscribe.EventResult<TransactionTerminatedEvent> terminatedEventEventResult = new SipSubscribe.EventResult<>(transactionTerminatedEvent);
+
+// sipSubscribe.getErrorSubscribe(header.getCallId()).response(terminatedEventEventResult);
}
@Override
public void processDialogTerminated(DialogTerminatedEvent dialogTerminatedEvent) {
CallIdHeader callId = dialogTerminatedEvent.getDialog().getCallId();
- System.out.println("processDialogTerminated:::::" + callId);
}
--
Gitblit v1.8.0