1c/OpenIntegrations
1
0
Fork 0
mirror of https://github.com/Bayselonarrend/OpenIntegrations.git synced 2026-06-20 09:19:27 +02:00
Code Issues Releases Activity

gRPC: Доработка подтверждения отправки

Browse Source
This commit is contained in:
Anton Titovets
2025-12-20 22:02:22 +03:00
parent fb81d92da9
commit eece669df8
5 changed files with 157 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/addins/grpc/dependencies.log
+6 -6
View File
@@ -1,10 +1,10 @@
"MAIN ---"
linux-vdso.so.1 (0x00007ffe0856a000)
libm.so.6 => /lib64/libm.so.6 (0x00007eb1fa200000)
libpthread.so.0 => /lib64/libpthread.so.0 (0x00007eb1f9e00000)
libc.so.6 => /lib64/libc.so.6 (0x00007eb1f9a00000)
libdl.so.2 => /lib64/libdl.so.2 (0x00007eb1f9600000)
/lib64/ld-linux-x86-64.so.2 (0x00007eb1fac00000)
linux-vdso.so.1 (0x00007ffdb21f3000)
libm.so.6 => /lib64/libm.so.6 (0x00007ff6c7600000)
libpthread.so.0 => /lib64/libpthread.so.0 (0x00007ff6c7200000)
libc.so.6 => /lib64/libc.so.6 (0x00007ff6c6e00000)
libdl.so.2 => /lib64/libdl.so.2 (0x00007ff6c6a00000)
/lib64/ld-linux-x86-64.so.2 (0x00007ff6c8000000)
GLIBC_2.2.5
GLIBC_2.12
GLIBC_2.3
src/addins/grpc/src/stream_manager.rs
+63 -22
View File
@@ -1,10 +1,15 @@
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};
use tokio::sync::{mpsc, oneshot, Mutex};
use tokio::task::AbortHandle;
use prost_reflect::{DynamicMessage, MessageDescriptor};
use uuid::Uuid;
pub struct MessageWithAck {
pub message: DynamicMessage,
pub ack: oneshot::Sender<()>,
}
#[derive(Clone, Debug)]
pub enum StreamType {
ServerStream,
@@ -15,8 +20,10 @@ pub enum StreamType {
pub struct StreamInfo {
pub stream_id: String,
pub stream_type: StreamType,
pub sender: Option<mpsc::UnboundedSender<DynamicMessage>>,
pub receiver: Arc<Mutex<Option<mpsc::UnboundedReceiver<DynamicMessage>>>>,
pub sender: Option<mpsc::Sender<DynamicMessage>>,
#[allow(dead_code)]
pub ack_sender: Option<mpsc::Sender<MessageWithAck>>,
pub receiver: Arc<Mutex<Option<mpsc::Receiver<DynamicMessage>>>>,
pub input_descriptor: Option<MessageDescriptor>,
pub output_descriptor: Option<MessageDescriptor>,
pub timeout_ms: Option<u64>,
@@ -25,7 +32,7 @@ pub struct StreamInfo {
impl StreamInfo {
pub fn new_server_stream(
receiver: mpsc::UnboundedReceiver<DynamicMessage>,
receiver: mpsc::Receiver<DynamicMessage>,
output_descriptor: MessageDescriptor,
timeout_ms: Option<u64>,
) -> Self {
@@ -33,6 +40,7 @@ impl StreamInfo {
stream_id: Uuid::new_v4().to_string(),
stream_type: StreamType::ServerStream,
sender: None,
ack_sender: None,
receiver: Arc::new(Mutex::new(Some(receiver))),
input_descriptor: None,
output_descriptor: Some(output_descriptor),
@@ -42,8 +50,9 @@ impl StreamInfo {
}
pub fn new_client_stream(
sender: mpsc::UnboundedSender<DynamicMessage>,
receiver: mpsc::UnboundedReceiver<DynamicMessage>,
sender: mpsc::Sender<DynamicMessage>,
ack_sender: mpsc::Sender<MessageWithAck>,
receiver: mpsc::Receiver<DynamicMessage>,
input_descriptor: MessageDescriptor,
output_descriptor: MessageDescriptor,
timeout_ms: Option<u64>,
@@ -52,6 +61,7 @@ impl StreamInfo {
stream_id: Uuid::new_v4().to_string(),
stream_type: StreamType::ClientStream,
sender: Some(sender),
ack_sender: Some(ack_sender),
receiver: Arc::new(Mutex::new(Some(receiver))),
input_descriptor: Some(input_descriptor),
output_descriptor: Some(output_descriptor),
@@ -61,8 +71,9 @@ impl StreamInfo {
}
pub fn new_bidi_stream(
sender: mpsc::UnboundedSender<DynamicMessage>,
receiver: mpsc::UnboundedReceiver<DynamicMessage>,
sender: mpsc::Sender<DynamicMessage>,
ack_sender: mpsc::Sender<MessageWithAck>,
receiver: mpsc::Receiver<DynamicMessage>,
input_descriptor: MessageDescriptor,
output_descriptor: MessageDescriptor,
timeout_ms: Option<u64>,
@@ -71,6 +82,7 @@ impl StreamInfo {
stream_id: Uuid::new_v4().to_string(),
stream_type: StreamType::BidiStream,
sender: Some(sender),
ack_sender: Some(ack_sender),
receiver: Arc::new(Mutex::new(Some(receiver))),
input_descriptor: Some(input_descriptor),
output_descriptor: Some(output_descriptor),
@@ -115,23 +127,53 @@ impl StreamManager {
let stream = self.get_stream(stream_id).await
.ok_or_else(|| format!("Stream '{}' not found", stream_id))?;
let sender = {
let (ack_sender, timeout_ms) = {
let stream_info = stream.lock().await;
stream_info.sender.as_ref()
let ack_sender = stream_info.ack_sender.as_ref()
.ok_or_else(|| format!("Stream '{}' does not support sending", stream_id))?
.clone()
.clone();
(ack_sender, stream_info.timeout_ms)
};
// Проверяем, не закрыт ли канал
if sender.is_closed() {
return Err(format!("Stream '{}' is closed", stream_id));
if ack_sender.is_closed() {
return Err("Closed".to_string());
}
// Пытаемся отправить
sender.send(message).map_err(|_| {
format!("Stream '{}' is closed", stream_id)
})
// Создаем oneshot канал для подтверждения
let (ack_tx, ack_rx) = oneshot::channel();
let msg_with_ack = MessageWithAck {
message,
ack: ack_tx,
};
// Отправляем сообщение
if let Some(timeout) = timeout_ms {
let duration = std::time::Duration::from_millis(timeout);
// Отправляем сообщение с таймаутом
tokio::time::timeout(duration, ack_sender.send(msg_with_ack))
.await
.map_err(|_| "Timeout")?
.map_err(|_| "Closed".to_string())?;
// Ждем подтверждения с таймаутом
tokio::time::timeout(duration, ack_rx)
.await
.map_err(|_| "Timeout")?
.map_err(|_| "Closed".to_string())?;
} else {
// Отправляем сообщение
ack_sender.send(msg_with_ack).await
.map_err(|_| "Closed".to_string())?;
// Ждем подтверждения
ack_rx.await
.map_err(|_| "Closed".to_string())?;
}
Ok(())
}
pub async fn receive_message(&self, stream_id: &str) -> Result<Option<DynamicMessage>, String> {
@@ -153,10 +195,10 @@ impl StreamManager {
Ok(Some(message)) => Ok(Some(message)),
Ok(None) => {
drop(receiver_guard);
Ok(None)
Err("Closed".to_string())
},
Err(_) => {
Err("Timeout waiting for message".to_string())
Err("Timeout".to_string())
},
}
} else {
@@ -228,8 +270,7 @@ impl StreamManager {
for (_, stream_arc) in streams.iter() {
let mut stream_info = stream_arc.lock().await;
drop(stream_info.sender.take());
// Отменяем фоновую задачу
if let Some(handle) = &stream_info.task_handle {
handle.abort();
}
src/addins/grpc/src/streaming_caller.rs
+57 -18
View File
@@ -5,13 +5,44 @@ use tonic::transport::Channel;
use tonic::{Request, Streaming};
use prost_reflect::{DescriptorPool, DynamicMessage, MethodDescriptor};
use prost::Message;
use tokio::sync::mpsc;
use tonic::codegen::tokio_stream;
use tokio::sync::{mpsc, oneshot};
use common_binary::vault::BinaryVault;
use crate::message_converter::{json_to_dynamic_message, dynamic_message_to_json};
use crate::grpc_caller::{apply_metadata, create_request_message};
use futures::StreamExt;
use crate::stream_manager::{StreamInfo, StreamManager};
use crate::stream_manager::{StreamInfo, StreamManager, MessageWithAck};
use std::pin::Pin;
use std::task::{Context, Poll};
use futures::Stream;
// Stream wrapper который отправляет подтверждение после каждого poll
struct AckStream {
receiver: mpsc::Receiver<MessageWithAck>,
current_ack: Option<oneshot::Sender<()>>,
}
impl Stream for AckStream {
type Item = Vec<u8>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
// Отправляем подтверждение для предыдущего сообщения
if let Some(ack) = self.current_ack.take() {
let _ = ack.send(());
}
// Получаем следующее сообщение
match self.receiver.poll_recv(cx) {
Poll::Ready(Some(msg_with_ack)) => {
let bytes = msg_with_ack.message.encode_to_vec();
self.current_ack = Some(msg_with_ack.ack);
Poll::Ready(Some(bytes))
}
Poll::Ready(None) => {
Poll::Ready(None)
}
Poll::Pending => Poll::Pending,
}
}
}
#[derive(serde::Deserialize)]
pub struct StreamCallParams {
@@ -55,7 +86,7 @@ pub async fn start_server_stream(
.map_err(|e| format!("gRPC server stream failed: {}", e))?
.into_inner();
let (tx, rx) = mpsc::unbounded_channel();
let (tx, rx) = mpsc::channel(1);
let handler_abort_handle = spawn_response_handler(response, method_desc.output(), tx);
@@ -76,12 +107,15 @@ pub async fn start_client_stream(
) -> Result<String, String> {
let method_desc = get_method_descriptor(descriptor_pool, &params.service, &params.method)?;
let (tx, rx) = mpsc::unbounded_channel::<DynamicMessage>();
let (tx, _rx) = mpsc::channel::<DynamicMessage>(1);
let (ack_tx, ack_rx) = mpsc::channel::<MessageWithAck>(1);
let byte_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx)
.map(|msg| msg.encode_to_vec());
let ack_stream = AckStream {
receiver: ack_rx,
current_ack: None,
};
let mut grpc_request = Request::new(byte_stream);
let mut grpc_request = Request::new(ack_stream);
apply_metadata(&mut grpc_request, metadata)?;
if let Some(timeout_ms) = params.timeout_ms {
@@ -95,7 +129,7 @@ pub async fn start_client_stream(
client.ready().await
.map_err(|e| format!("Failed to ready client: {}", e))?;
let (response_tx, response_rx) = mpsc::unbounded_channel();
let (response_tx, response_rx) = mpsc::channel(1);
let output_desc = method_desc.output();
let task = tokio::spawn(async move {
@@ -112,7 +146,7 @@ pub async fn start_client_stream(
let bytes = response.into_inner();
match DynamicMessage::decode(output_desc, bytes.as_ref()) {
Ok(message) => {
let _ = response_tx.send(message);
let _ = response_tx.send(message).await;
}
Err(e) => {
eprintln!("Failed to decode client stream response: {}", e);
@@ -127,6 +161,7 @@ pub async fn start_client_stream(
let mut stream_info = StreamInfo::new_client_stream(
tx,
ack_tx,
response_rx,
method_desc.input(),
method_desc.output(),
@@ -148,13 +183,16 @@ pub async fn start_bidi_stream(
) -> Result<String, String> {
let method_desc = get_method_descriptor(descriptor_pool, &params.service, &params.method)?;
let (tx_send, rx_send) = mpsc::unbounded_channel::<DynamicMessage>();
let (tx_recv, rx_recv) = mpsc::unbounded_channel();
let (tx_send, _rx_send) = mpsc::channel::<DynamicMessage>(1);
let (ack_tx, ack_rx) = mpsc::channel::<MessageWithAck>(1);
let (tx_recv, rx_recv) = mpsc::channel(1);
let byte_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx_send)
.map(|msg| msg.encode_to_vec());
let ack_stream = AckStream {
receiver: ack_rx,
current_ack: None,
};
let mut grpc_request = Request::new(byte_stream);
let mut grpc_request = Request::new(ack_stream);
apply_metadata(&mut grpc_request, metadata)?;
if let Some(timeout_ms) = params.timeout_ms {
@@ -182,6 +220,7 @@ pub async fn start_bidi_stream(
let mut stream_info = StreamInfo::new_bidi_stream(
tx_send,
ack_tx,
rx_recv,
method_desc.input(),
method_desc.output(),
@@ -196,12 +235,12 @@ pub async fn start_bidi_stream(
fn spawn_response_handler(
mut response: Streaming<Vec<u8>>,
output_descriptor: prost_reflect::MessageDescriptor,
sender: mpsc::UnboundedSender<DynamicMessage>,
sender: mpsc::Sender<DynamicMessage>,
) -> tokio::task::AbortHandle {
let task = tokio::spawn(async move {
while let Ok(Some(bytes)) = response.message().await {
if let Ok(message) = DynamicMessage::decode(output_descriptor.clone(), bytes.as_ref()) {
if sender.send(message).is_err() {
if sender.send(message).await.is_err() {
break;
}
}
src/ru/OPI/src/CommonModules/OPI_GRPC/Module.bsl
-21
View File
@@ -535,27 +535,6 @@
КонецФункции
Функция ПолучитьСтатусПотока(Знач Соединение, Знач IDПотока) Экспорт
Если Не ЭтоКоннектор(Соединение) Тогда
СоответствиеОшибки = Новый Соответствие;
СоответствиеОшибки.Вставить("result", Ложь);
СоответствиеОшибки.Вставить("error" , "Implicit connection opening is senseless in this case");
Возврат СоответствиеОшибки;
КонецЕсли;
OPI_ПреобразованиеТипов.ПолучитьСтроку(IDПотока);
Результат = Соединение.GetStreamStatus(IDПотока);
Результат = OPI_Инструменты.JsonВСтруктуру(Результат);
Возврат Результат;
КонецФункции
#КонецОбласти
#КонецОбласти
src/ru/OPI/src/CommonModules/OPI_Тесты/Module.bsl
Vendored
+31 -34
View File
@@ -26316,12 +26316,22 @@
МассивСообщений = Новый Массив;
ТекущееСообщение = Новый Структура("hasMore", Истина);
Пока ТекущееСообщение["hasMore"] Цикл
Пока Истина Цикл
ТекущееСообщение = OPI_GRPC.ПолучитьСообщение(Соединение, IDПотока);
Если Не ТекущееСообщение["result"] Тогда
ВызватьИсключение Результат["error"];
Ошибка = ТекущееСообщение["error"];
Если Ошибка = "Timeout" Тогда
Продолжить;
ИначеЕсли Ошибка = "Closed" Тогда
Прервать;
Иначе
ВызватьИсключение Ошибка;
КонецЕсли;
Иначе
ДанныеСообщения = ТекущееСообщение["message"];
@@ -26388,38 +26398,25 @@
Счетчик = 0;
Пока Истина Цикл
СтатусПотока = OPI_GRPC.ПолучитьСтатусПотока(Соединение, IDПотока);
Если Не СтатусПотока["result"] Тогда
ВызватьИсключение СтатусПотока["error"];
КонецЕсли;
ИнформацияОПотоке = СтатусПотока["data"];
Если Не ИнформацияОПотоке["isActive"] Тогда
Прервать;
КонецЕсли;
МожноПринимать = ИнформацияОПотоке["canReceive"];
МожноОтправлять = ИнформацияОПотоке["canSend"];
Если МожноПринимать Тогда
Результат = OPI_GRPC.ПолучитьСообщение(Соединение, IDПотока);
Прервать;
КонецЕсли;
Если МожноОтправлять Тогда
ТекущаяОтправка = OPI_GRPC.ОтправитьСообщение(Соединение, IDПотока, Данные);
Если Не ТекущаяОтправка["result"] Тогда
ВызватьИсключение ТекущаяОтправка["error"] + СтрШаблон("(отправлено сообщений: %1)", Счетчик);
КонецЕсли;
Счетчик = Счетчик + 1;
КонецЕсли;
ТекущаяОтправка = OPI_GRPC.ОтправитьСообщение(Соединение, IDПотока, Данные);
Если Не ТекущаяОтправка["result"] Тогда
Ошибка = ТекущаяОтправка["error"];
Если Ошибка = "Timeout" Тогда
Продолжить;
ИначеЕсли Ошибка = "Closed" Тогда
Результат = OPI_GRPC.ПолучитьСообщение(Соединение, IDПотока);
Прервать;
Иначе
ВызватьИсключение Ошибка + СтрШаблон(" (отправлено сообщений %1)", Счетчик);
КонецЕсли;
КонецЕсли;
Счетчик = Счетчик + 1;
КонецЦикла;