Преглед на файлове

1.增加协调器客户端的通道主动关闭接口

2.增加视频设备断开通知,将所有连接该设备的客户端全部切断
master
SmallChi(Koike) преди 4 години
родител
ревизия
0aae0e872a
променени са 4 файла, в които са добавени 57 реда и са изтрити 9 реда
  1. +3
    -3
      src/JT1078.Gateway.Tests/JT1078.Gateway.TestNormalHosting/Program.cs
  2. +12
    -0
      src/JT1078.Gateway/JT1078CoordinatorHttpClient.cs
  3. +16
    -2
      src/JT1078.Gateway/Jobs/JT1078SessionNoticeJob.cs
  4. +26
    -4
      src/JT1078.Gateway/Sessions/JT1078HttpSessionManager.cs

+ 3
- 3
src/JT1078.Gateway.Tests/JT1078.Gateway.TestNormalHosting/Program.cs Целия файл

@@ -30,10 +30,10 @@ namespace JT1078.Gateway.TestNormalHosting
services.AddSingleton(typeof(ILogger<>), typeof(Logger<>));
//使用内存队列实现会话通知
services.AddJT1078Gateway(hostContext.Configuration)
.AddHttp()
.AddUdp()
.AddTcp()
//.AddCoordinatorHttpClient()
.AddUdp()
.AddHttp()
.AddCoordinatorHttpClient()
.AddNormal()
.AddMsgProducer()
.AddMsgConsumer();


+ 12
- 0
src/JT1078.Gateway/JT1078CoordinatorHttpClient.cs Целия файл

@@ -43,5 +43,17 @@ namespace JT1078.Gateway
{
await httpClient.PostAsync($"{endpoint}/heartbeat", new StringContent(content));
}

/// <summary>
/// 发送设备号和通道给协调器中
/// </summary>
/// <param name="terminalPhoneNo"></param>
/// <param name="channelNo"></param>
public async ValueTask ChannelClose(string terminalPhoneNo,int channelNo)
{
//todo:通过自维护,当协调重启导致集群内网关未关闭的情况下,通过轮询的方式再去调用
string json = $"{{\"TerminalPhoneNo\":\"{terminalPhoneNo}\",\"ChannelNo\":\"{channelNo}\"}}";
await httpClient.PostAsync($"{endpoint}/ChannelClose", new StringContent(json));
}
}
}

+ 16
- 2
src/JT1078.Gateway/Jobs/JT1078SessionNoticeJob.cs Целия файл

@@ -1,8 +1,11 @@
using JT1078.Gateway.Services;
using JT1078.Gateway.Abstractions;
using JT1078.Gateway.Services;
using JT1078.Gateway.Sessions;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Text;
using System.Text.Json;
using System.Threading;
@@ -15,12 +18,15 @@ namespace JT1078.Gateway.Jobs
{
private readonly ILogger logger;
private JT1078SessionNoticeService SessionNoticeService;
private JT1078HttpSessionManager HttpSessionManager;
public JT1078SessionNoticeJob(
JT1078SessionNoticeService sessionNoticeService,
ILoggerFactory loggerFactory)
ILoggerFactory loggerFactory,
[AllowNull]JT1078HttpSessionManager jT1078HttpSessionManager=null)
{
logger = loggerFactory.CreateLogger<JT1078SessionNoticeJob>();
SessionNoticeService = sessionNoticeService;
HttpSessionManager = jT1078HttpSessionManager;
}

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -34,6 +40,14 @@ namespace JT1078.Gateway.Jobs
{
logger.LogInformation($"[Notice]:{notice.TerminalPhoneNo}-{notice.ProtocolType}-{notice.SessionType}");
}
if(JT1078GatewayConstants.SessionOffline== notice.SessionType)
{
if (HttpSessionManager != null)
{
//当1078设备主动断开的情况下,需要关闭所有再观看的连接
HttpSessionManager.TryRemoveBySim(notice.TerminalPhoneNo);
}
}
}
}
catch


+ 26
- 4
src/JT1078.Gateway/Sessions/JT1078HttpSessionManager.cs Целия файл

@@ -48,18 +48,40 @@ namespace JT1078.Gateway.Sessions
{

}
finally
}
}

public async void TryRemoveBySim(string sim)
{
var keys=Sessions.Where(f => f.Value.Sim == sim).Select(s => s.Key);
foreach(var key in keys)
{
if (Sessions.TryRemove(key, out JT1078HttpContext session))
{
//todo:session close notice
try
{
if (session.IsWebSocket)
{
await session.WebSocketClose("close");
}
else
{

await session.HttpClose();
}
}
catch (Exception)
{

}
}
}
}
}

private void remove(string sessionId)
{
if (Sessions.TryRemove(sessionId, out JT1078HttpContext session))
{
//todo:session close notice
}
}



Зареждане…
Отказ
Запис