////////////////////////////////////////////////////////////////////////////////
// cdc1cДроссель — общий модуль (флаги: Сервер, Внешнее соединение).
//
// Ограничивает размер пакета SelectChanges. Платформа формирует весь пакет в памяти рабочего
// процесса и параметры запроса ($top/$select/$filter) к SelectChanges игнорирует — проверено:
// ответ с ними байт в байт совпадает с ответом без них. Значит ограничить пакет со стороны
// клиента (onecdc) нечем, и делать это приходится здесь, регистрацией изменений.
//
// Схема — два плана обмена:
//   A (накопитель) — авторегистрация РАЗРЕШЕНА, состав = все реплицируемые объекты.
//                    SelectChanges по нему не вызывается никогда; мы только читаем его таблицы
//                    <Объект>.Изменения запросом (объекты при этом не читаются — это дёшево).
//   B (отдача)     — тот же состав, авторегистрация ЗАПРЕЩЕНА для всех объектов. Единственный
//                    источник регистраций в нём — эта процедура. Из узла B читает onecdc.
//
// Двух узлов одного плана недостаточно: авторегистрация ставит регистрацию на все узлы плана,
// и очередь-отдача наполнялась бы сама, мимо дросселя. Признак «Авторегистрация» задаётся на
// паре план × объект состава, поэтому очередь с выключенной авторегистрацией — это отдельный план.
//
// Это минимальный работающий пример: один лимит на всю очередь и круговой обход состава.
// Если нужна логика тоньше — приоритеты объектов, лимит в строках или мегабайтах вместо
// объектов, отдельные лимиты для тяжёлых документов, — она пишется здесь же; для onecdc
// ничего не меняется, он как читал SelectChanges с узла B, так и читает.
//
// Вызывается регламентным заданием: cdc1cДроссель.ПеренестиПорцию (см. комментарий в конце файла).
////////////////////////////////////////////////////////////////////////////////

#Область ПараметрыНастройки

// План-накопитель: тот, что заполняется авторегистрацией (обычно уже существующий план обмена).
Функция ИмяПланаНакопителя()
	Возврат "ДляODATA";
КонецФункции

// План-отдача: с запрещённой авторегистрацией, из него читает onecdc (ONECDC_EXCHANGE_NAME).
Функция ИмяПланаОтдачи()
	Возврат "ДляODATAОтдача";
КонецФункции

// Сколько единиц регистрации допустимо держать в очереди-отдаче. Единица — ссылка
// (документ/справочник) или регистратор (набор движений регистра), поэтому лимит считается
// в объектах, а не в мегабайтах: 100 реализаций с ТЧ на 5000 строк дадут пакет в сотни
// мегабайт при любом «небольшом» лимите. Подбирается по размеру пакета, который onecdc пишет
// в лог каждым циклом.
Функция ЛимитОчереди()
	Возврат 500;
КонецФункции

#КонецОбласти

#Область ПрограммныйИнтерфейс

// Добивает очередь-отдачу до лимита регистрациями из плана-накопителя.
// Метод регламентного задания; параллельных запусков одного задания платформа не допускает.
Процедура ПеренестиПорцию() Экспорт

	ПланОтдачи = Метаданные.ПланыОбмена.Найти(ИмяПланаОтдачи());
	Если ПланОтдачи = Неопределено Тогда
		ВызватьИсключение "onecdc дроссель: не найден план обмена """ + ИмяПланаОтдачи() + """";
	КонецЕсли;

	УзелНакопителя = УзелПлана(ИмяПланаНакопителя());
	УзелОтдачи = УзелПлана(ИмяПланаОтдачи());

	// Очередь-отдача не обязана быть пустой: добираем ровно до лимита. Если onecdc уже выбрал
	// пакет, но не подтвердил его, добавленное сюда получит следующий номер сообщения и уедет
	// отдельным пакетом — подтверждение снимает регистрацию только по своему номеру.
	Остаток = ЛимитОчереди() - КоличествоРегистраций(УзелОтдачи, ПланОтдачи, ЛимитОчереди());
	Если Остаток <= 0 Тогда
		Возврат;
	КонецЕсли;

	Объекты = ОбъектыОбхода(ПланОтдачи);
	Если Объекты.Количество() = 0 Тогда
		Возврат;
	КонецЕсли;

	Перенесено = Новый Соответствие;
	Позиция = ПозицияОбхода(Объекты.Количество());
	Обработано = 0;
	Всего = 0;

	// Полный круг по составу, начиная с сохранённой позиции: без round-robin хвостовые объекты
	// ждали бы, пока опустеет голова очереди, а голова пополняется авторегистрацией постоянно.
	Пока Обработано < Объекты.Количество() И Остаток > 0 Цикл
		ОбъектМетаданных = Объекты[Позиция];

		Взято = ПеренестиОбъект(ОбъектМетаданных, УзелНакопителя, УзелОтдачи, Остаток);
		Если Взято > 0 Тогда
			Перенесено.Вставить(ОбъектМетаданных.ПолноеИмя(), Взято);
			Остаток = Остаток - Взято;
			Всего = Всего + Взято;
		КонецЕсли;

		Позиция = ?(Позиция + 1 = Объекты.Количество(), 0, Позиция + 1);
		Обработано = Обработано + 1;
	КонецЦикла;

	СохранитьПозициюОбхода(Позиция);

	Если Всего > 0 Тогда
		Описание = "";
		Для Каждого КлючЗначение Из Перенесено Цикл
			Описание = Описание + Символы.ПС + "    " + КлючЗначение.Ключ + ": " + КлючЗначение.Значение;
		КонецЦикла;
		Лог("Перенесено " + Всего + " единиц регистрации:" + Описание);
	КонецЕсли;

КонецПроцедуры

#КонецОбласти

#Область СлужебныеПроцедурыИФункции

// Единственный узел плана, кроме ЭтотУзел. Узлов должно быть ровно два (ЭтотУзел и наш):
// неоднозначность здесь означала бы, что порция уедет не в ту очередь, — лучше упасть.
Функция УзелПлана(ИмяПлана)

	Запрос = Новый Запрос;
	Запрос.Текст =
		"ВЫБРАТЬ
		|	Т.Ссылка КАК Ссылка
		|ИЗ
		|	ПланОбмена." + ИмяПлана + " КАК Т
		|ГДЕ
		|	Т.Ссылка <> &ЭтотУзел";
	Запрос.УстановитьПараметр("ЭтотУзел", ПланыОбмена[ИмяПлана].ЭтотУзел());
	Узлы = Запрос.Выполнить().Выгрузить();

	Если Узлы.Количество() <> 1 Тогда
		ВызватьИсключение "onecdc дроссель: в плане обмена """ + ИмяПлана + """ должен быть ровно "
			+ "один узел кроме ЭтотУзел, найдено — " + Узлы.Количество();
	КонецЕсли;

	Возврат Узлы[0].Ссылка;

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

// Сколько единиц регистрации сейчас в узле. Считаем по составу плана и прекращаем, как только
// набрали Порог: точное число сверх лимита никому не нужно, а таблицы изменений бывают большими.
Функция КоличествоРегистраций(Узел, ПланМетаданных, Порог)

	Количество = 0;

	Для Каждого ЭлементСостава Из ПланМетаданных.Состав Цикл
		Запрос = Новый Запрос;
		Запрос.Текст =
			"ВЫБРАТЬ
			|	КОЛИЧЕСТВО(*) КАК Количество
			|ИЗ
			|	" + ЭлементСостава.Метаданные.ПолноеИмя() + ".Изменения КАК Т
			|ГДЕ
			|	Т.Узел = &Узел";
		Запрос.УстановитьПараметр("Узел", Узел);

		Выборка = Запрос.Выполнить().Выбрать();
		Если Выборка.Следующий() Тогда
			Количество = Количество + Выборка.Количество;
		КонецЕсли;

		Если Количество >= Порог Тогда
			Прервать;
		КонецЕсли;
	КонецЦикла;

	Возврат Количество;

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

// Объекты, которые дроссель умеет переносить: пересечение составов обоих планов минус то,
// что не поддерживает ни обмен, ни onecdc. Регистрация объекта, не входящего в состав плана
// отдачи, — ошибка платформы, поэтому состав B проверяется явно.
Функция ОбъектыОбхода(ПланОтдачи)

	Объекты = Новый Массив;

	Для Каждого ЭлементСостава Из Метаданные.ПланыОбмена[ИмяПланаНакопителя()].Состав Цикл
		ОбъектМетаданных = ЭлементСостава.Метаданные;

		Если ПланОтдачи.Состав.Найти(ОбъектМетаданных) = Неопределено Тогда
			Продолжить;
		КонецЕсли;

		// Независимый регистр сведений: регистрация ведётся по набору измерений, регистратора нет.
		// Обменом такие регистры не поддерживаются (и onecdc их не читает) — как и в обработке.
		Если Метаданные.РегистрыСведений.Содержит(ОбъектМетаданных)
			И ОбъектМетаданных.РежимЗаписи = Метаданные.СвойстваОбъектов.РежимЗаписиРегистра.Независимый Тогда
			Продолжить;
		КонецЕсли;

		Если ЭтоРегистрСРегистратором(ОбъектМетаданных)
			Или Метаданные.Константы.Содержит(ОбъектМетаданных)
			Или ЭтоСсылочныйОбъект(ОбъектМетаданных) Тогда
			Объекты.Добавить(ОбъектМетаданных);
		КонецЕсли;
	КонецЦикла;

	Возврат Объекты;

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

// Переносит до Лимит единиц регистрации по одному объекту. Возвращает сколько перенесено.
Функция ПеренестиОбъект(ОбъектМетаданных, УзелНакопителя, УзелОтдачи, Лимит)

	ПолноеИмя = ОбъектМетаданных.ПолноеИмя();
	ЭтоРегистр = ЭтоРегистрСРегистратором(ОбъектМетаданных);
	ЭтоКонстанта = Метаданные.Константы.Содержит(ОбъектМетаданных);

	// У таблицы изменений константы нет колонки объекта — регистрация одна на узел.
	// Порядок выборки не задаём намеренно: сортировка таблицы изменений на сотнях тысяч строк
	// стоит дороже, чем даёт, а порядок доставки внутри пакета для CDC роли не играет.
	Запрос = Новый Запрос;
	Если ЭтоКонстанта Тогда
		Запрос.Текст =
			"ВЫБРАТЬ ПЕРВЫЕ 1
			|	1 КАК Поле
			|ИЗ
			|	" + ПолноеИмя + ".Изменения КАК Т
			|ГДЕ
			|	Т.Узел = &Узел";
	Иначе
		Запрос.Текст =
			"ВЫБРАТЬ ПЕРВЫЕ " + Формат(Лимит, "ЧГ=0") + "
			|	Т." + ?(ЭтоРегистр, "Регистратор", "Ссылка") + " КАК Объект
			|ИЗ
			|	" + ПолноеИмя + ".Изменения КАК Т
			|ГДЕ
			|	Т.Узел = &Узел";
	КонецЕсли;
	Запрос.УстановитьПараметр("Узел", УзелНакопителя);
	Выборка = Запрос.Выполнить().Выбрать();

	Менеджер = ?(ЭтоРегистр, ПолучитьМенеджерРегистра(ОбъектМетаданных), Неопределено);
	Счётчик = 0;

	// Регистрация в отдаче и снятие с накопителя — одной транзакцией: иначе отказ между ними
	// оставил бы изменение не зарегистрированным нигде, то есть потерянным навсегда. Обратный
	// порядок (сначала снять) той же проблемой и обернулся бы. Транзакция короткая — она
	// ограничена порцией, ради чего лимит и вводится.
	НачатьТранзакцию();
	Попытка

		Если ЭтоКонстанта Тогда
			Если Выборка.Следующий() Тогда
				ПланыОбмена.ЗарегистрироватьИзменения(УзелОтдачи, ОбъектМетаданных);
				ПланыОбмена.УдалитьРегистрациюИзменений(УзелНакопителя, ОбъектМетаданных);
				Счётчик = 1;
			КонецЕсли;
		Иначе
			Пока Выборка.Следующий() Цикл
				Если ЭтоРегистр Тогда
					// Набор записей регистратора — одна единица регистрации и одна entry в пакете,
					// сами записи читать не нужно, достаточно отбора.
					Регистрируемый = Менеджер.СоздатьНаборЗаписей();
					Регистрируемый.Отбор.Регистратор.Установить(Выборка.Объект);
				Иначе
					// Ссылка может указывать на уже удалённый объект: удаление тоже регистрируется,
					// и в пакет оно попадёт как объект удаления.
					Регистрируемый = Выборка.Объект;
				КонецЕсли;

				ПланыОбмена.ЗарегистрироватьИзменения(УзелОтдачи, Регистрируемый);
				ПланыОбмена.УдалитьРегистрациюИзменений(УзелНакопителя, Регистрируемый);
				Счётчик = Счётчик + 1;
			КонецЦикла;
		КонецЕсли;

		ЗафиксироватьТранзакцию();

	Исключение
		ОтменитьТранзакцию();
		// Сбой по одному объекту не должен останавливать перенос остальных: следующий запуск
		// возьмёт эту порцию снова — в накопителе она осталась нетронутой.
		Лог("Ошибка переноса """ + ПолноеИмя + """: " + ОписаниеОшибки(),
			УровеньЖурналаРегистрации.Ошибка);
		Возврат 0;
	КонецПопытки;

	Возврат Счётчик;

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

// Позиция круга хранится в общих настройках, а не в константе: чтобы дроссель ставился
// расширением, не требуя новых объектов конфигурации.
Функция ПозицияОбхода(Количество)
	Позиция = ХранилищеОбщихНастроек.Загрузить("cdc1cДроссель", "ПозицияОбхода");
	Если ТипЗнч(Позиция) <> Тип("Число") Или Позиция >= Количество Или Позиция < 0 Тогда
		Возврат 0;
	КонецЕсли;
	Возврат Позиция;
КонецФункции

Процедура СохранитьПозициюОбхода(Позиция)
	ХранилищеОбщихНастроек.Сохранить("cdc1cДроссель", "ПозицияОбхода", Позиция);
КонецПроцедуры

Функция ЭтоРегистрСРегистратором(ОбъектМетаданных)
	Возврат Метаданные.РегистрыНакопления.Содержит(ОбъектМетаданных)
		Или Метаданные.РегистрыБухгалтерии.Содержит(ОбъектМетаданных)
		Или Метаданные.РегистрыРасчета.Содержит(ОбъектМетаданных)
		Или (Метаданные.РегистрыСведений.Содержит(ОбъектМетаданных)
			И ОбъектМетаданных.РежимЗаписи <> Метаданные.СвойстваОбъектов.РежимЗаписиРегистра.Независимый);
КонецФункции

Функция ЭтоСсылочныйОбъект(ОбъектМетаданных)
	Возврат Метаданные.Справочники.Содержит(ОбъектМетаданных)
		Или Метаданные.Документы.Содержит(ОбъектМетаданных)
		Или Метаданные.ПланыСчетов.Содержит(ОбъектМетаданных)
		Или Метаданные.ПланыВидовХарактеристик.Содержит(ОбъектМетаданных)
		Или Метаданные.ПланыВидовРасчета.Содержит(ОбъектМетаданных)
		Или Метаданные.БизнесПроцессы.Содержит(ОбъектМетаданных)
		Или Метаданные.Задачи.Содержит(ОбъектМетаданных);
КонецФункции

Функция ПолучитьМенеджерРегистра(ОбъектМетаданных)
	ИмяРегистра = ОбъектМетаданных.Имя;
	Если Метаданные.РегистрыНакопления.Содержит(ОбъектМетаданных) Тогда
		Возврат РегистрыНакопления[ИмяРегистра];
	ИначеЕсли Метаданные.РегистрыСведений.Содержит(ОбъектМетаданных) Тогда
		Возврат РегистрыСведений[ИмяРегистра];
	ИначеЕсли Метаданные.РегистрыБухгалтерии.Содержит(ОбъектМетаданных) Тогда
		Возврат РегистрыБухгалтерии[ИмяРегистра];
	ИначеЕсли Метаданные.РегистрыРасчета.Содержит(ОбъектМетаданных) Тогда
		Возврат РегистрыРасчета[ИмяРегистра];
	КонецЕсли;
	Возврат Неопределено;
КонецФункции

Процедура Лог(Текст, Уровень = Неопределено)
	ЗаписьЖурналаРегистрации("onecdc.Дроссель",
		?(Уровень = Неопределено, УровеньЖурналаРегистрации.Информация, Уровень),
		, , Текст);
КонецПроцедуры

#КонецОбласти

////////////////////////////////////////////////////////////////////////////////
// Установка
//
// 1. План обмена B (например, «ДляODATAОтдача»): состав — те же объекты, что в плане-накопителе,
//    у КАЖДОГО объекта состава Авторегистрация = Запретить. Создать в нём один узел (кроме
//    ЭтотУзел) — из него будет читать onecdc.
// 2. Общий модуль «cdc1cДроссель» с этим кодом: флаги Сервер и Внешнее соединение,
//    Клиент/Вызов сервера — нет.
// 3. Регламентное задание (например, «cdc1cПереносПорцииИзменений»), имя метода —
//    cdc1cДроссель.ПеренестиПорцию, расписание — раз в 10–30 секунд. Задание, заставшее
//    полную очередь-отдачу, сразу выходит, поэтому частый запуск дёшев.
// 4. В onecdc переключить репликатор на план B: ONECDC_EXCHANGE_NAME = имя плана отдачи,
//    ONECDC_QUEUE_GUID = Ref_Key его узла (при пустом queue_guid onecdc выведет список узлов
//    в лог — см. ChangeReader._raise_no_queue_guid).
// 5. План-накопитель после переключения НЕ читать через SelectChanges: его очередь теперь
//    расходует дроссель, и параллельное чтение уведёт изменения мимо onecdc.
////////////////////////////////////////////////////////////////////////////////
