////////////////////////////////////////////////////////////////////////////////
// onecdcДроссель — общий модуль (флаги: Сервер, Внешнее соединение).
//
// Ограничивает размер пакета SelectChanges. 

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

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

// Точка входа регламентного задания: имя метода — onecdcДроссель.ЗапуститьПеренос1.
// Настройка вся здесь, никаких констант и параметров задания.
//
Процедура ЗапуститьПеренос1() Экспорт

	// План-накопитель: тот, что заполняется авторегистрацией (обычно уже существующий план обмена).
	Накопитель = "Для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 Тогда
		Описание = "";
		Для Каждого КлючЗначение Из Перенесено Цикл
			Описание = Описание + Символы.ПС + "    " + КлючЗначение.Ключ + ": " + КлючЗначение.Значение;
		КонецЦикла;
		Лог("Перенесено в """ + Очередь.Имя + """ " + Всего + " единиц регистрации:" + Описание);
	КонецЕсли;

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

// Очереди с их метаданными и узлами, в порядке настройки. Проверяем всё здесь, до первой
// транзакции: опечатка в имени плана должна ронять задание, а не наполнять не ту очередь.
Функция ПланыОтдачи(ИмяНакопителя, ИменаОтдач)

	Если ИменаОтдач.Количество() = 0 Тогда
		ВызватьИсключение "onecdc дроссель: не задан ни один план-отдача";
	КонецЕсли;

	Очереди = Новый Массив;
	Имена = Новый Соответствие;

	Для Каждого ИмяПлана Из ИменаОтдач Цикл

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

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

		// Пустое пересечение составов — всегда ошибка настройки: чаще всего в точке входа указан
		// не тот накопитель. Молчать нельзя, иначе задание годами «работает», ничего не перенося.
		Если ОбъектыОбхода(ИмяНакопителя, ПланМетаданных).Количество() = 0 Тогда
			ВызватьИсключение "onecdc дроссель: состав плана-отдачи """ + ИмяПлана + """ не пересекается "
				+ "с составом накопителя """ + ИмяНакопителя + """ — переносить нечего";
		КонецЕсли;

		Очередь = Новый Структура("Имя, Метаданные, Узел", ИмяПлана, ПланМетаданных, УзелПлана(ИмяПлана));
		Очереди.Добавить(Очередь);

	КонецЦикла;

	Возврат Очереди;

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

// Узлы остальных очередей: зарегистрированное в них в порцию не берётся.
Функция УзлыДругихОчередей(Очереди, Текущая)

	Узлы = Новый Массив;
	Для Каждого Очередь Из Очереди Цикл
		Если Очередь.Узел <> Текущая.Узел Тогда
			Узлы.Добавить(Очередь.Узел);
		КонецЕсли;
	КонецЦикла;

	Возврат Узлы;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

	ПолноеИмя = ОбъектМетаданных.ПолноеИмя();
	ЭтоРегистр = ЭтоРегистрСРегистратором(ОбъектМетаданных);
	ЭтоКонстанта = Метаданные.Константы.Содержит(ОбъектМетаданных);
	ЕстьДругиеОчереди = ДругиеУзлы.Количество() > 0;

	// У таблицы изменений константы нет колонки объекта: регистрация одна на узел, поэтому
	// «стоит в другой очереди» проверяется по узлу целиком, а не условием в запросе порции.
	Если ЭтоКонстанта И ЕстьДругиеОчереди И ЗарегистрированВУзлах(ПолноеИмя, ДругиеУзлы) Тогда
		Возврат 0;
	КонецЕсли;

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

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

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

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

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

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

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

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

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

// Есть ли регистрация объекта хоть в одном из узлов. Только для констант — см. ПеренестиОбъект.
Функция ЗарегистрированВУзлах(ПолноеИмя, Узлы)

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

	Возврат Не Запрос.Выполнить().Пустой();

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

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

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

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

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

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

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

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