Шина данных на Apache Kafka

// Процедура - Сообщение для отправки в узел
//
// Параметры:
//  Сообщение		 - Структура	- Сообщение для отправки в сервисную шину предприятия см. СервиснаяШинаПредприятияКлиентСервер.Сообщение() 
//  ОбъектИзменений	 - СправочникОбъект - Объект справочника изменения которого отправляются 
//  Узел			 - ПланОбменаОбъект - Узел для которого формируется сообщение
//
Процедура СообщениеДляОтправкиВУзел(Сообщение, ОбъектИзменений, Узел = Неопределено) Экспорт
	
	Сообщение.Key = XMLСтрока(ОбъектИзменений.Ссылка);
	
	Если ТипЗнч(ОбъектИзменений) = Тип("УдалениеОбъекта") Тогда Возврат; КонецЕсли;
	
	СтруктураОбмена = СтруктураОбмена();
	
	ОписаниеЗаполнения = ?(ОбъектИзменений.ЭтоГруппа, ОписаниеЗаполненияГруппы(), ОписаниеЗаполнения());
	
	УправлениеМастерДаннымиОбменДанными.ЗаполнитьСтруктуруОбменаИзОбъекта(СтруктураОбмена, ОбъектИзменений, ОписаниеЗаполнения);
	
	ЗаписьJSON = Новый ЗаписьJSON;
	ЗаписьJSON.УстановитьСтроку();
	ЗаписатьJSON(ЗаписьJSON, СтруктураОбмена);
	
	Сообщение.Value = ЗаписьJSON.Закрыть();		
	
КонецПроцедуры

// Отправка сообщений
Функция ПолучитьСообщенияДляСервиснойШиныПредприятия(ПараметрыУзла, ПараметрыТопика, НомерСообщенияОбмена) Экспорт
	
	СервиснаяШинаПредприятия.ПолучитьПорциюНаОтправку(ПараметрыУзла.Ссылка, "Справочник.Номенклатура", НомерСообщенияОбмена);
	
	Сообщения = Новый Массив;
	
	ДанныеИзменений = СобратьДанныеПоИзменениям(Узел, НомерСообщенияОбмена);
	
	Для Каждого ОбъектИзменений Из ДанныеИзменений Цикл
		
		Сообщение = СервиснаяШинаПредприятияКлиентСервер.Сообщение();
		Сообщение.Key = XMLСтрока(ОбъектИзменений.Ссылка);
		
		Сообщения.Добавить(Сообщение);
		  
		Если ОбъектИзменений.УдалениеОбъекта_ Тогда Продолжить; КонецЕсли;
		
		СтруктураОбмена = ?(ОбъектИзменений.ЭтоПапка, СтруктураОбменаПапка(), СтруктураОбмена());
		УправлениеМастерДаннымиОбменДанными.ЗаполнитьСтруктуруОбменаИзОбъекта(СтруктураОбмена, ОбъектИзменений, ОписаниеЗаполнения());
		
		ЗаписьJSON = Новый ЗаписьJSON;
		ЗаписьJSON.УстановитьСтроку(Новый ПараметрыЗаписиJSON(ПереносСтрокJSON.Нет));
		ЗаписатьJSON(ЗаписьJSON, СтруктураОбмена);
		
		Сообщение.Value = ЗаписьJSON.Закрыть();		
		Сообщение.Headers.Добавить(Новый Структура("AvailableCountries", СтруктураОбмена["AvailableCountries"]));
		
	КонецЦикла;

	Возврат Сообщения;
	
КонецФункции

// Функция - Разобрать входящее сообщение
//
// Параметры:
//  Сообщение	 - Структура - Входящее сообщение из сервисной шины предприятия (Apache Kafka)
//		* Headers - Массив из Структур:
//			** Ключ - Строка - Ключ заголовка 
//			** Значение - Произвольный - Значение заголовка
//		* Key - Строка - Ключ сообщения
//		* Offset - Число - Смещение сообщения
//		* Partition - Число - Номер раздела
//		* Timestamp - Число - Отпечаток времени
//		* Topic - Строка - Имя топика
//		* Value - Строка - Тело сообщения
// 
// Возвращаемое значение:
//	Булево - Истина при успешном результате разбора сообщения
//
Функция РазобратьВходящееСообщение(Сообщение, Узел = Неопределено) Экспорт
	
	Если НРег(Сообщение.Value) = "null" Тогда Возврат Истина; КонецЕсли;
	
	ЧтениеJSON = Новый ЧтениеJSON;
	ЧтениеJSON.УстановитьСтроку(Сообщение.Value);
	СтруктураОбмена = ПрочитатьJSON(ЧтениеJSON, Истина);
	
	СсылкаИзменяемого = ПолучитьСсылку(Новый УникальныйИдентификатор(Сообщение.Key)); 
	
	ИзменяемыйОбъект = СсылкаИзменяемого.ПолучитьОбъект();
	Если ИзменяемыйОбъект = Неопределено Тогда
		ИзменяемыйОбъект = ?(СтруктураОбмена["IsFolder"] = Истина, СоздатьГруппу(), СоздатьЭлемент());
	КонецЕсли;
	
	ОписаниеЗаполнения = ?(ИзменяемыйОбъект.ЭтоГруппа, ОписаниеЗаполненияГруппы(), ОписаниеЗаполнения());
	
	УправлениеМастерДаннымиОбменДанными.ЗаполнитьОбъектИзСтруктурыОбмена(ИзменяемыйОбъект, СтруктураОбмена, ОписаниеЗаполнения);
	
	Если ЗначениеЗаполнено(Узел) Тогда
		Попытка 
			ИзменяемыйОбъект.ОбменДанными.Отправитель = Узел;
		Исключение
		КонецПопытки;
	КонецЕсли;
	
	ИзменяемыйОбъект.ОбменДанными.Загрузка = Истина;
	ИзменяемыйОбъект.Записать();
	
	Возврат Истина;

КонецФункции

//----- Разбор
Функция РазобратьВходящиеСообщенияСервиснойШины(Сообщения, Узел) Экспорт
	
	ТаблицыВходныхДанных = ТаблицыВходныхДанных();
	ОсновныеДанные = ТаблицыВходныхДанных.ОсновныеДанные;
	
	Для Каждого Сообщение Из Сообщения Цикл 
		// TODO требуется определить поведение: null значения - УдалениеОбъекта,  при удалении объекта в базе источнике изменений 
		Если НРег(Сообщение.Value) = "null" Тогда Продолжить; КонецЕсли;
		
		Для Каждого Заголовок Из Сообщение.Headers Цикл
			КлючЗаголовка = ОбщегоНазначенияКлиентСервер.СвойствоСтруктуры(Заголовок, "Key");
			Если КлючЗаголовка = "DataCountry" Тогда
				КодСтраныДанных = ОбщегоНазначенияКлиентСервер.СвойствоСтруктуры(Заголовок, "Value");
			КонецЕсли;
		КонецЦикла;
		
		Если ЗначениеЗаполнено(КодСтраныДанных) Тогда 
			СтранаДанных = Справочники.КлассификаторСтранМира.НайтиПоРеквизиту("КодАльфа2", КодСтраныДанных);
		КонецЕсли;
		
		ЧтениеJSON = Новый ЧтениеJSON;
		ЧтениеJSON.УстановитьСтроку(Сообщение.Value);
		ОбъектXDTO = ФабрикаXDTO.ПрочитатьJSON(ЧтениеJSON, ТипОбъектаXDTO());
		
		СтрокаТз = ОсновныеДанные.Добавить();
		СтрокаТз.ОтпечатокВремени	= Сообщение.Offset;
		СтрокаТз.Представления 		= ТаблицыВходныхДанных.Представления.СкопироватьКолонки();
		СтрокаТз.Родители			= ТаблицыВходныхДанных.Родители.СкопироватьКолонки();
		СтрокаТз.ДоступныеСтраны	= ТаблицыВходныхДанных.ДоступныеСтраны.СкопироватьКолонки();
		СтрокаТз.СертификатыРСТ		= ТаблицыВходныхДанных.СертификатыРСТ.СкопироватьКолонки();	
		СтрокаТз.СтранаДанных		= СтранаДанных;	
		
		ОписаниеСтруктуры = ?(ОбъектXDTO.IsFolder, ОписаниеСтруктурыXDTOДляОбменаГруппа(), ОписаниеСтруктурыXDTOДляОбмена());
		УправлениеМастерДаннымиОбменДанными.ЗаполнитьПоСоответствиюРекурсивно(СтрокаТз, ОбъектXDTO, ОписаниеСтруктуры);
		
		// Заполнение доп. данные по номенклатуре
		УправлениеМастерДаннымиОбменДанными.ЗаполнитьПоСоответствиюРекурсивно(СтрокаТз, ОбъектXDTO, ОписаниеСтруктурыXDTOСтатусовНоменклатуры());
		// Поля могут быть Булево или неопределено, заполняем вручную
		ЗаполнитьЗначенияСвойств(СтрокаТз, ОбъектXDTO, "Available, doc_traceability, has_asc_warranty"); 
	
	КонецЦикла;
	
	ОбъектыНаОбновление = ПолучитьОбъектыНаОбновление(ОсновныеДанные);
	Для Каждого Объект Из ОбъектыНаОбновление Цикл
		ВТранзакции = ТипЗнч(Объект) = Тип("Массив");
		Если ВТранзакции Тогда
			
			НачатьТранзакцию();
			Попытка
				Для Каждого ОбновляемыйОбъект Из Объект Цикл
					ОбновляемыйОбъект.ОбменДанными.Загрузка = Истина;
					ОбновляемыйОбъект.Записать();
		        КонецЦикла;
				ЗафиксироватьТранзакцию();
			Исключение
				ОтменитьТранзакцию();
				ВызватьИсключение;
			КонецПопытки;
			
		Иначе
			Объект.ОбменДанными.Загрузка = Истина;
			Объект.Записать();
		КонецЕсли;
	КонецЦикла;
	
	Возврат Истина;
		
КонецФункции