Савичев Дмитрий Викторович. Разработка расширяемого сетевого сервиса межпротокольного взаимодействия для высоконагруженных телематических систем с асинхронной сервис-ориентированной архитектурой

Аннотация
Тема: «Разработка расширяемого сетевого сервиса межпротокольного
взаимодействия для высоконагруженных телематических систем с асинхронной
сервис-ориентированной архитектурой»
Объем выпускной квалификационной работы страниц 116 страниц из
которых приложение занимает 49 страниц, на которых размещены 13 рисунков и
2 таблицы. Приложение содержит фрагменты программного кода. При
написании диплома использовалось 28 источников.
Ключевые
слова:
телематика,
Интернет
вещей,
межпротокольное
взаимодействие, сервис-ориентированная архитектура, подключение устройств.
Целью данной работы является разработка расширяемого сервиса,
способного обеспечить межпротокольное взаимодействие для телематических
систем.
Для достижения цели в первой главе проведено изучение сопутствующей
литературы.
Выбранная
литература
связана
с
системным
анализом,
телематическими системами в том числе транспортными, Интернетом вещей,
сервис-ориентированной архитектурой и сетевым взаимодействием.
Вторая глава содержит исследование предметной области. Рассмотрена
транспортная телематика и интернет вещей.
В третьей главе описывается процесс проектирование сервиса, включая
определения списка задач, которые должен выполнять сервис. Описан способ
перимения разработанного сервиса,
рассмотрены подсистемы сервиса и их
функции.
Четвертая заключает в себе информацию по программной реализации
сервиса. Описание способов расширения сервиса.
Разработанный сервис может как приемная часть для телематических
систем. Его особенностью является возможность расширения и настройки
работы сервиса без модификации основного приложения.
4
Содержание
Введение
6
1 Теоретические основы разработки сетевого сервиса межпротокольного
взаимодействия для высоконагруженных телематических систем
7
1.1 Обзор и анализ литературных источников
7
1.1.1 Системное проектирование
10
1.1.2 Транспортные телематические системы и их использование
11
1.1.3 Применение Интернета вещей
16
1.2 Библиографическая справка
17
1.2.1 Синхронность и асинхронность
18
1.2.2 Сервис-ориентированная архитектура
20
1.2.3 Boost.Asio и его применение
22
1.2.4 Сопрограммы
24
1.2.5 Межпротокольное взаимодействие
27
2
Исследование
области
применения
сервиса
межпротокольного
взаимодействия
29
2.1 Предметная область
29
2.1.1 Транспортная телематика
30
2.1.2 Проблемы интернета вещей
31
2.2 Поиск и изучение готовых решений в области применения сервиса
37
2.2.1 Интернет вещей
37
2.2.2 Транспортная телематика
39
2.2.3 Выводы об области применения
42
3 Проектирование расширяемого сетевого сервиса межпротокольного
взаимодействия для высоконагруженных телематических систем с
асинхронной сервис-ориентированной архитектурой
43
3.1 Задачи поставленные перед разработкой сервиса
43
3.2 Применение разработанного сервиса в платформе для Интернета
вещей Rightech IoT Cloud
44
5
3.3 Разработка расширяемого сетевого сервиса
межпротокольного
взаимодействия
45
3.4 Архитектура сервиса
50
3.4.1 Ядро системы
51
3.4.2 Подсистема обработки данных от объектов
52
3.4.3 Подсистема взаимодействия с другими сервисами
54
3.5 Применение сопрограмм для реализации асинхронности
55
3.6 Безопасность
56
4 Реализация сервиса RicGate
59
4.1 Описание инструментария
59
4.2 Описание реализации
59
4.3 Добавление плагинов
60
4.3.1 Входной плагин
61
4.3.2 Выходной плагин
62
4.3.3 Сборка сервиса
62
Заключение
63
Список использованной литературы
64
Приложение А – Фрагменты кода
68
Удостоверяющий лист № 156100/п
117
Информационно-поисковая характеристика документа на электронном
носителе
118
6
Введение
В
настоящее
время
тематические
системы
развиваются
очень
стремительно и спрос на подобные системы огромен. Однако необходимо
помнить, что в процесс разработки таких систем необходимо учитывать большое
количество различных особенностей.
Телематические системы имеют очень сложную структуру, состоящую из
множества незаменимых частей. Незаменимым элементом подобных систем
является
подсистема,
связанная
с
организацией
межпротокольного
взаимодействия. Подобные сервисы должны обеспечивать первичную обработку
данных и реализацию правильного и безопасного взаимодействия в процессе
обмена данными в телематической системе. Сложность подобных систем может
разниться в зависимости от определенных требований к конкретной системе.
Может
меняться
как
величина
нагрузки
(количество
единовременных
соединений между участниками обмена данными), так и степень обработки
данных (вплоть до полного разбора данных от объектов контроля и
регулированием процесса отправки команд на них).
В
процессе
разработки
сетевого
сервиса
межпротокольного
взаимодействия важную роль занимает построение его архитектуры. К данной
задаче можно подходить с разных сторон, однако обязательными этапами ее
решения является:
– изучение теоритической базы по построению и реализации систем с
асинхронной сервис-ориентированной архитектурой;
– поиск и исследование реальных сервисов с изучением их основных
достоинств и недостатков;
– разработка архитектуры cервиса межпротокольного взаимодействия на
основе данных полученных на предыдущих этапах;
– разработка
сервиса,
отвечающего
определены на следующих этапах.
требованиям,
которые
будут
7
1
Теоретические
основы
межпротокольного
разработки
взаимодействия
для
сетевого
сервиса
высоконагруженных
телематических систем
1.1
Обзор и анализ литературных источников
С целью теоретического анализа и построения процесса разработки
расширяемого
сетевого
высоконагруженных
сервиса
межпротокольного
телематических
систем
с
взаимодействия
асинхронной
для
сервис-
ориентированной архитектурой будут исследованы материалы из различных
источников, как зарубежных, так и отечественных авторов. Цель исследования
будет сопоставлена с результатами, отраженными в рассматриваемых работах.
Тема выпускной квалификационной работы напрямую связана с работой
высоконагруженных телематических систем, поэтому для того чтобы описать
предметную область и в полной мере изучить проблематику, необходимо
изучить основные термины и определения.
Под телематикой, исходя из греческого перевода этого слова(один из
вариантов перевода – «удаленное воздействие»), понимается удаленное
управлние объектами. Нередко этот термин жестко связывают с транспортной
телематикой, хотя это лишь раздел телематики.Обычно выделяют следующие
разделы телематики:
– транспортная телематика;
– автоматизация зданий (организация производства);
– телематика услуг (бизнес, коммерция, логистика, правительство);
– телематика здоровья;
– телематика безопасности и др.
Данное разделение условно, но позволяет приблизительно оценить
масштабы данной области. Чтобы дать определение телематической системы
рассмотрим транспортную телематику.
8
Телематической
обеспечивающую
представление
системой
называется
автоматизированный
потребителям
данных
информационную
сбор,
о
обработку,
местоположении
систему,
передачу
и
и
состоянии
транспортных средств, а также информации, получаемой на основе этих данных,
в целях эффективного и безопасного использования транспортных средств
различного назначения и принадлежности [1, c.4]. Кроме того, в последнее время
к функциям телематических систем относят и возможность управления
отдельными элементами транспортных средств удаленно, то есть с помощью
телематической системы. Данное определение позволяет охарактеризовать одно
из способов применения сервиса созданного по результатам данной работы.
В своей работе В.В. Комаров [1] проводит исследование телематических
систем в нескольких странах, что позволяет использовать этот опыт в процессе
диссертационной работы. В нашей стране данный вид систем до недавнего
времени был развит слабо, и для раскрытия потенциала этой отрасли
необходимо провести большую работу, как связанную с правовыми аспектами,
так и с процессом разработки и применения таких систем.
Последнее время автомобильный сегмент микроэлектроники является,
пожалуй,
одним
полупроводниковой
из
наиболее
динамично
промышленности.
В
развивающихся
направлений
автомобильной
электронике
появляются принципиально новые направления, самым молодым и наиболее
перспективным из которых является телематика. Термин «телематика» является
производным от двух: телекоммуникации и информатика.
Автомобили
перестают быть просто транспортным средством в классическом понимании.
Перспективы рынка телематики выглядят довольно многообещающе. Все
крупнейшие производители автомобилей и автомобильной электроники в той
или иной степени вовлечены в процесс разработки телематических систем [2].
Задачи первых транспортных телематических систем сводились к простому
мониторингу движения транспортных средств. В наше время возможности таких
систем существенно расширились.
9
В настоящее время, говоря о телематических системах нельзя не
упоминуть Интернет вещей. С развитием и распространением данной отрасли
изменились требования к телематическим системам. Существует много
определений термина Интернет вещей (Internet of Things, IoT). Одно из
определений интернета вещей, которое отражает основную идею Интернета
вещей говорит, что IoT – это концепция пространства, в котором все из
аналогового и цифрового миров может быть совмещено [3, c.31]. Фактически это
означает, что Интернет вещей – это не просто множество различных приборов и
датчиков, объединенных между собой проводными и беспроводными каналами
связи и подключенных к сети Интернет, а это более тесная интеграция реального
и виртуального миров, в котором общение производится между людьми и
устройствами. Системы, а в отрасли IoT их чаще называют платформами,
предназначенные для сбора данных и управлениями различными объектами
должны обладать широким спектром возможностей. Такие платформы должны
иметь широкую область применения и разнообразие в процессе настройки
системы. То есть необходимо возможность контроля разнообразных объектов и
для различных целей.
Телематические системы очень сложны по своей структуре, и должны
учитывать множество различных факторов, чтобы обеспечивать безопасность
использования транспортных средств.
Как следует из приведенных определений, телематические системы
являются
информационными
системами,
следовательно,
на
них
распространяются общие теоретические и прикладные результаты, полученные в
ходе широкого круга исследований и разработок по проблемам анализа и
синтеза систем в рамках таких междисциплинарных научных направлений, как
теория систем (см. [4]), системный анализ (см. [5]) и др. Системный анализ, в
целом, позволяет более обоснованно с единообразных методологических
позиций, эффективно ставить и решать многочисленные задачи в сфере
формирования и совершенствования информационных систем[6, c.42].
10
1.1.1 Системное проектирование
В процессе создания подобных систем, системное проектирование
является
фундаментом
для
обеспечения
функциональной
адекватности
требованиям всего жизненного цикла сложных систем. От полноты и
тщательности системного проектирования зависят эффективность реализации
функций системы и степень удовлетворения ожиданий и требований заказчика и
пользователей. Антов А.В. в своей работе [5] в последовательности выработки и
подготовки к реализации этих требований выделяются три крупных этапа:
– обследование, системный анализ существующей системы и выявление
ее недостатков;
– обобщение результатов системного анализа и создание предварительной
концепции новой или модернизированной системы и ее программных средств;
– разработка проекта системы, определяющего и конкретизирующего
цель, назначение и методы ее дальнейшего детального проектирования и всего
жизненного цикла.
Системное проектирование способно остановить нерентабельное развитие
проектов систем и избежать крупных затрат заказчикам и разработчикам. В то
же
время
на
базе
рекомендуемых
при
проектировании
методов,
инструментальных средств и стандартов может и должен быть подготовлен и
обеспечен длительный, эффективный жизненный цикл и совершенствование
множества версий высококачественных систем и их компонентов при
реализации на различных аппаратных и операционных платформах.
Конечный
результат
системного
проектирования
должен
также
положительно отражаться на системах обеспечения качества, безопасности и
защиты
системы,
на
рационально
организованных
коллективах
квалифицированных специалистов, способных обеспечить весь жизненный цикл
системы [7, c.182].
11
1.1.2 Транспортные телематические системы и их использование
Область применения и цели создания телематических систем могут быть
разнообразны. Комаров В.В. в своей работе [8], выделяет следующие возможные
задачи таких систем:
– взимание платы за проезд транспортных средств по дорогам (участкам
уличнодорожной сети), на которых установлена такая плата;
– дистанционный контроль и надзор при осуществлении перевозок
опасных и ценных грузов;
– контроль времени работы и отдыха водителей и пресечение нарушений
установленно го режима;
– обеспечение транспортной безопасности ТС, находящихся в рейсе;
– управление системами (группами) ТС, например, автотранспортом
оперативных служб, дорожноэксплуатационных и коммунальных организаций,
грузовых и пассажирских автопредприятий, таксопарками и т. д.;
– формирование оптимальных (рациональных) маршрутов движения ТС,
сообщивших пункты своего назначения, и прогнозирование продолжительности
поездки;
– оптимальное управление дорожным движением;
– информирование пользователей телематических систем (водителей и
пассажиров);
– экстренное реагирование на аварийные ситуации;
– розыск угнанных ТС;
– использование поступающих в телематические системамы данных о
транспортных средствах для решения задач правоохранительных органов.
Данный список может меняться и дополняться в зависимости от
выбранной направленности и области применения для разрабатываемой
телематической системы.
12
Для взаимодействия между транспортным средством и телематической
системой необходима установка специального оборудования, телематических
терминалов. Телематический терминал – устройство для приема и обработки
сигналов навигационных спутников. Он предназначен для установки на
транспортное средство как дополнительное устройство, регистрирующие
местоположение, скорость и направление движения транспортного средства,
например по системе ГЛОНАСС/GPS. Абонентский телематический терминал
может получать данные с датчиков о состоянии объекта, обеспечивая связь с
телематическим сервером, передавая управляющие сигналы на исполнительные
устройства [9, с.22]. В настоящие время такие терминалы иногда встраиваются в
ТС при производстве, но пока это скорее исключение.
Для
представления
некоторого
итога
по
проблемам,
решаемых
транспортными телематическими системами можно использовать верхний
уровень классификации проблем, решаемых телематическими системами в
трактовках, используемых ИСО, США и Европейским Союзом (таблица 1).
Отсутствие прямого соответствия по отдельным элементам связано с разным
подходом
к
классификации.
Данное
сравнение
позволяет
оценить
направленность создания транспортных телематических систем в разных
регионах. На следующих уровнях классификации наблюдается большее
соответствие.
Анализ проблем, представленных в таблице 1, и их дальнейшей
детализации в [8,10,11] и других материалах показывает, что предметные
области практически всех проблем включают движение всех транспортных
средств (ТС) либо отдельных их категорий на всей территории, обслуживаемой
системой. Данная таблица позволяет оценить единство и отличия классификации
проблем, решаемых телематическими ситемами с точки зрения стандартов места
их применения. Для сравнения были взяты стандарты РФ, США и Европейского
союза.
13
Таблица
1
–
Верхний
уровень
классификации
проблем,
решаемых
телематическими системами
Сервисный домен по
ГОСТ Р ИСО 14813–
1 — 2011 [10]
1
1 Информирование
участников
движения
2 Управление
дорожным
движением и
действий к его
участникам
3 Конструкция
транспортных
средств
4 Коммерческие
перевозки
5 Общественный
транспорт
6 Чрезвычайные
ситуации
7 Электронные
платежи на
транспорте
8 Персональная
безопасность,
связанная с
дорожным
транспортом
9 Погодные условия и
состояние
окружающей среды
Группа пользовательских
сервисов Национальной
архитектуры ИТС США
[11]
2
–
Управление
перемещениями и
дорожным движением
(Travel and Traffic
Management)
Усовершенствованные
системы активной
безопасности (Advanced
Vehicle Safety Systems)
Деятельность грузового
транспорта (Commercial
Vehicle Operations)
Управление общественным
транспортом (Public
Transportation Management)
–
Электронные платежи
(Electronic Payment)
Фрагмент ИТС (Part of
ITS) в Европейской
рамочной архитектуре
ИТС [8]
3
Помощь
путешественнику
(Traveller Assistance)
Управление дорожным
движением (Traffic
Management)
Системы на
транспортном средстве
(In-Vehicle Systems)
Управление грузами и
грузоперевозками (Freight
and Fleet Management)
Управление
общественным
транспортом (Public
Transport Management)
–
–
Сбор
электронныхплатежей
(Electronic Fee Collection)
–
–
–
14
Продолжение таблицы 1
1
10 Катастрофы и
чрезвычайные
ситуации
2
Управление в
чрезвычайных ситуациях
(Emergency Management)
11 Национальная
безопасность
12 Управление
данными ИТС
13 –
–
14 –
В
состав
Управление информацией
(Information Management)
Управление дорожными
работами (Maintenance and
Construction Management)
–
информации,
получаемой
3
Оповещение и реакция
на чрезвычайные
ситуации (Emergency
Notification and
Response)
Правоприменение (Law
Enforcement)
–
–
Поддержка
кооперативныхсистем
(Cooperative Systems)
системой,
должны
входить
местоположение и скорость движения транспортных средств (ТС). Кроме этого в
зависимости от решаемой задачи могут быть включены и другие параметры
(контроль уровня топлива, расчет пройденного пути, оценка состояния
элементов транспортного средства).
Навигация
является
основной
и
наиболее
распространенной
телематической услугой, которая одной из первых была внедрена в автомобиль.
Как правило, это связано с использованием приемника глобальной системы
позиционирования (GPS) и интерактивной картографической базы данных для
предоставления водителю услуги четкого и подробного ведения по маршруту с
помощью визуальных и голосовых подсказок о маневрах.
Автомобильные телематические системы безопасности сочетают в себе
сенсорные и радиочастотные технологии для обнаружения и предотвращения
экстренных ситуаций во время вождения.
Удаленная диагностика позволяет заблаговременно определять возможные
электрические или механические проблемы транспортного средства, тем самым
15
способствуя предотвращению поломки автомобиля, а также минимизации
рисков, связанных со здоровьем водителя и сохранностью грузов. Передача
показаний автомобильных датчиков и других диагностических данных
напрямую из автомобиля в сервис-центр позволяет проводить своевременное
техническое обслуживание и ремонт транспортного средства без необходимости
планировать такие работы. Удаленная диагностика позволяет экономить время и
сокращать эксплуатационные расходы [2].
Рисунок 1 – Схема взаимодействия транспортного средства и
телематической системы
На рис. 1 представлена простая схема работы телематической системы, на
примере управления и контроля транспортного средства. На данной схеме видно,
что данные от транспорта (точнее от установленного на транспорт устройства)
передаются на сервер с помощью сотовой связи (один из основных вариантов в
данной области).
В последнее время все большую популярность и применение получают
телематические системы с возможностью удаленного управления автомобилем.
16
Здесь имеется в виду контроль над отдельными частями транспортного средства.
Например, подобные системы могут обладать следующими функциями:
– блокировка/открытие дверей;
– блокировка/открытие багажника;
– постановка на общую охрану;
– автозапуск автомобиля и т.д.
Подобный функционал может обеспечиваться лишь на некоторых моделях
автомобилях, число которых растет с каждым годом.
Однако появление возможности удаленного управления транспортным
средством создает ряд угроз, которые необходимо учитывать при разработке
телематических систем.
В условиях дальнейшего углубления компьютеризации транспортных
средств наблюдатели выражают обеспокоенность ростом числа уязвимостей
автомобилей. Насыщенность современных автомобилей электроникой таит в
себе угрозы. Исследователям уже удалось дистанционными методами взять под
контроль тормозную систему, организовать прослушивание разговоров внутри
салона и проделать еще много других угроз [13].
Данные
собранные
получаемые
от
контролируемых
объектов
формируются по специальным протоколам, которые зависят от установленного
оборудования.
Их существует большое количество. Поэтому необходима
добавлять или изменять реализованные в телематической системе протоколы.
Под протоколом обычно понимается набор правил и соглашений, используемых
при передаче данных. Данное определение варьируется в зависимости от места
применения.
1.1.3 Применение Интернета вещей
Область
применения
Интернета
вещей
огромна.
IoT
решения
обеспечивают подключения устройств, сбор данных, извлечение аналитической
17
информации, управления объектами (в командном и автоматическом режимах) в
различных областях окружающей действительности:
– умный транспорт;
– энергетика;
– здравоохранение;
– умное производство;
– торговля;
– умные здания и многое др.
Уже повсеместно расширяется область их применения в нашей жизни. Но
с точки зрения системы IoT каждый объект должен абстрагируется от своего
предназначения. Поэтому платформы IoT имеет широкий уровень настройки для
работы с любыми объектами, контролирование которых необходимо.
1.2
Библиографическая справка
1.2.1 Синхронность и асинхронность
Основные схемы для построения сетевых приложений: синхронное и
асинхронное программирование.
Рисунок 2 – Синхронный подход
18
В
синхронном
программировании
все
операции
выполняются
в
последовательном порядке, такие как чтение (запрос) из сокета, а затем написать
(ответ) в сокет. Графическое представление синхронного подхода представлена
на рис. 2. Каждая из операций является блокирующей.
Так как операции
блокирующие, то чтобы не прерывать основную программу, в то время как
происходит считывание или запись в сокет, обычно создаются один или
несколько потоков, которые имеют дело с сокетами ввода/вывода. Таким
образом, синхронные сервер/клиенты, как правило, многопоточны [14, c. 44].
Для синхронных приложений характерна простота (как в написании кода,
так и в понимании концепции) и последовательность в реализации. К
недостаткам зачастую относят бесполезную затрату ресурсов и большое время
простоя. При
этом данные недостатки проявляются более выражено при
увеличении числа потоков.
Асинхронное программирование чрезвычайно отличается от синхронного
программирования. Асинхронные программы управляются событиями. Когда
начинается операция, неизвестно, когда она закончится; предпочтительнее
выхвать callback функцию, которая будет вызываться API с результатом
операции, когда операция завершится [15, с. 35]. Графическое представление
синхронного подхода представлена на рис. 3.
Рисунок 3 – Асинхронный подход
19
Использование
асинхронной
архитектуры
обусловлено
тем,
что
разрабатываемый сервис должен обеспечивать обработку большого объема
информации поступающих от большего числа источников. Оценка возможной
допустимой нагрузки и расчет необходимого её уровня будет рассмотрен в
следующих главах данной работы.
Термин асинхронный означает, что события (элементарные сигналы) не
совпадают с какими–то определенными моментами реального времени. Следует
иметь в виду, что асинхронная передача и несинхронизированная передача – это
не одно и то же. Несинхронизированную передачу данных чаще называют
мусором (garbage).
Асинхронная модель построена на очереди событий (event-loop). При
возникновении некоторого события (например, пришел запрос, выполнилось
считывание файла) оно помещается в конец очереди. Поток, который
обрабатывает эту очередь, берет событие с начала очереди, и выполняет
связанный с этим событием код. Пока очередь не пуста процессор будет занят
работой. Почти все операции неблокирующие. Блокирующие тоже имеются, но
их использование крайне не рекомендуется.
В разных источниках данная архитектура может называться по-разному.
Часто, это связано с тем, что область ее применении обширно и не всегда слово
архитектура уместно в конкретном случае.
Асинхронный подход характеризуется:
– асинхронная реализация;
– производительность;
– относительно простая параллелизация.
К недостаткам относят:
– передачу контекста;
– нелинейную структуру кода;
– усложненный процесс обработки ошибок;
– время жизни объектов;
20
– сложный процесс отладки кода.
Кроме того, с ростом количества операций сложность растет нелинейно
[17].
Главная особенность асинхронной архитектуры заключается в том, что
события обрабатываются независимо друг от друга. Асинхронное приложение с
неблокирующими операциями использует процессорное время эффективнее, но
более
сложно
при
проектировании.
Поэтому
асинхронная
архитектура
необходима при работе с высоконагруженными телематическими системами.
Событийный цикл в асинхронно событийной модели можно описать как
бесконечный цикл, который опрашивает «источники событий» (дескрипторы) на
предмет появления в них какого–нибудь «события». Опрос происходит с
помощью библиотеки «синхронного» ввода/вывода, который, при этом будет
являться «неблокирующим». То есть, во время очередного витка «событийного
цикла», наша система проходит последовательно по всем дескрипторам, и
пытается считать из них «события»: если таковые имеются, то они возвращаются
функцией чтения в нашу систему; если же никаких новых событий у
дескриптора нет, то он не станет «блокировать» и ждать появления «события», а
сразу же возвратит ответ: «новых событий нет». Это описание позволяет лучше
понять смысл асинхронной архитектуры.
«Асинхронная событийная модель» хорошо подойдёт там, где много
пользователей
одновременно
нагружающие процессор.
производят
какие-нибудь
действия,
не
Например, получают температуру с датчиков в
режиме «текущего времени». Что верно и для телематических систем, где в роли
пользователей выступают терминалы, установленные на транспортные средства.
1.2.2 Сервис-ориентированная архитектура
Сегодня наблюдается устойчивый рост интереса к концепции сервис–
ориентированной архитектуры (service-oriented architecture, SOA). Технология
21
Web-сервисов
—
это
набор
основанных
на
XML
спецификаций,
обеспечивающих универсальный метод технического описания сервисов и
взаимодействия с ними. Сами сервисы, реализованные в соответствии с этими
спецификациями, называют Web-сервисами. Возможно, именно технологии
Web-сервисов сыграли свою роль в наименовании сервис-ориентированной
архитектуры.
Сервис-ориентированная архитектура не является чем-то новым: ITотделы
компаний
успешно
создавали
и
развертывали
приложения,
поддерживающие сервис-ориентированную архитектуру, уже много лет –
задолго до появления XML и Web-сервисов.
Сервис–ориентированная архитектура – это не технология, а способ
проектирования и организации информационной архитектуры и бизнесфункциональности.
В самом общем виде сервис-ориентированная архитектура предполагает
наличие трех основных участников: поставщика сервиса, потребителя сервиса и
реестра сервисов. Взаимодействие участников выглядит достаточно просто:
поставщик сервиса регистрирует свои сервисы в реестре, а потребитель
обращается к реестру с запросом
Сервис-ориентированная архитектура "подталкивает" к использованию
альтернативных технологий и подходов (таких как обмен сообщениями) для
построения приложений посредством связывания сервисов, а не посредством
написания нового программного кода. В этом случае, при надлежащем
проектировании, применение сообщений позволяет компаниям своевременно
реагировать на изменение рыночных условий – "настраивать" процесс обмена
сообщениями, а не разрабатывать новые приложения [15].
Применение
сервис-ориентированной
архитектуры
для
создания
телематических систем позволяет упростить процесс внесения изменений в
систему после ее запуска.
Преимущества внедрения сервис-ориентироанной архитектуры:
22
– сокращение времени реализации проектов, или "времени выхода на
рынок";
– повышение производительности;
– более быстрая и менее дорогая интеграция приложений и интеграция
B2B [17].
Можно сказать, что SOA — это развитие известных принципов
структурного и объектно-ориентированного программирования: разработка
элементарных стандартных процедур и функций, годных к многократному
повторному использованию, но на новом витке. Особенностью SOA является то,
что она требует перехода с уровня разработки, каталогизации и типизации
отдельных детальных операций и алгоритмов на уровень бизнес-функций и
бизнес-процессов [17].
1.2.3 Boost.Asio и его применение
Boost.Asio это, большей частью, кросс-платформенная С++ библиотека для
программирования сетей и некоторых других низкоуровневых программ
ввода/вывода.
Есть много реализаций для решения сетевых задач, но Boost.Asio
превзошел их все; он был принят в Boost в 2005 и с тех пор был протестирован
большим количеством пользователей Boost, используется во многих проектах.
Boost.Asio успешно абстрагирует понятия input и output, которые работают не
только для работы в сети, но и для последовательных COM-портов, файлов и так
далее. Кроме этого вы можете делать input или output программирование
синхронным или асинхронным. Даная особенность этой библиотеки позволяет
сравнить
оба
варианта
решения
поставленной
задачи
(асинхронное и
асинхронное программирование).
Библиотека
является
переносимой,
работает
на
большинстве
операционных систем и хорошо масштабируется при более чем тысяче
23
одновременных подключений [16, c.33]. Это позволит обеспечить работы с
высоконагруженными сетями, что является необходимым условием в данной
работе.
Когда дело доходит до потоков в Boost.Asio, нам надо поговорить о
следующем:
– io_service: класс io_service является потоко-безопасным. Несколько
потоков могут вызвать io_service::run(). Чаще всего вы, вероятно, вызываете
io_service::run() из одного потока, так что функция ждет пока все блокирующие
асинхронные функции будут выполнены. Тем не менее вы можете вызывать
io_service::run() из нескольких потоков. Это блокирует все потоки, которые
будут вызывать io_service::run(). Все функции обратного вызов будут вызваны в
контекстах вех потоков, которые вызвали io_service::run(); это так же означает,
что если вы вызвали io_service::run(), только в одном потоке, то все функции
обратного вызова будут вызваны в контексте данного потока.
– socket: классы сокетов не являются потоко-безопасными. Таким образом
вам следует избегать таких ситуаций как читать из сокета в одном потоке, а
писать в него в другом (это рекомендуется вообще, не говоря уже о Boost.Asio).
– utility: Классы utility обычно не имеет смысла использовать в нескольких
потоках, они не потоко-безопасны. Большинство из них используются короткое
время, а затем выходят из области видимости.[16, с.77]
Кроме сетевого функционала Boost.Asio позволяет выполнять следующие
функции:
– таймер (некоторые операции ввода/вывода могут иметь временные
ограничения для завершения) Это можно применить это только к асинхронным
операциям (так как синхронные средства блокирования не имеют временных
ограничений;
– другие объекты ввода/вывода не относящиеся к сетям.
24
1.2.4 Сопрограммы
Сопрограмма (англ. coroutine) — компонент программы, обобщающий
понятие подпрограммы, который дополнительно поддерживает множество
входных точек (а не одну как подпрограмма), остановку и продолжение
выполнения с сохранением определённого положения.
Необходимость использования сопрограмм на практике естественным
образом возникает в случае, когда они связаны с алгоритмами ввода и
вывода,[18, c.221] что можно отнести и к данной работе, так необходимо
реализовать процесс приема и отправки информации. Основное отличие
сопрограмм от синхронной реализации состоит в том что нет блокирующих
операций.
К
недостаткам
асинхронного
программирования
относят
передачу
контекста, нелинейную структуру кода, усложненный процесс обработки
ошибок, время жизни объектов, сложный процесс отладки кода. Кроме того, с
ростом количества операций сложность растет нелинейно[21, с.25].Выходом
может стать использование сопрограмм. Сопрограммы позволяют объединить
достоинства и убрать недостатки, присущие обоим методам.
Использование сопрограмм для альтернативы асинхронной реализации не
ново. Их использование к примеру можно встретить в таком языке
программирования как Python. Однако в C++ данная абстракция еще не
включена в стандарт языка и неизвестно когда это произойдет (пока инициатива
включить
сопрограммы
в
стандарт
C++17
отклонена
комитетом
по
стандартизации).
Основное отличие сопрограмм от синхронной реализации состоит в том
что нет блокирующих операций. Особенностью применения абстракции
сопрограмм, если ее применять правильным образом для правильной проблемы,
то код не теряет ничего, но при этом может приобрести в производительности.
25
Иногда возникает необходимость в разработке алгоритмов сразу для
нескольких компьютерных конфигураций, одни из которых имеют больший
объем памяти, чем другие. В таких случаях можно написать программу в виде
набора сопрограмм и поставить число проходов в зависимость от размера
памяти: загрузить сразу столько сопрограмм, сколько поместится, а подачу
данных по недостающим связям обеспечить с помощью подпрограмм ввода или
вывода [18, с.224].
Рисунок 4 – Пример использования сопрограмм
На данной схеме (рис.4) представлен идеальный вариант работы с
сопрограммами, но такая реализация не будет надежно работать из-за
многопоточности, так не исключает состояние гонки. Решение, которое позволит
избавиться от данной ситуации, заключается в откладывание асинхронной
операции до выхода из сопрограммы (yield) [18, с.226]. Асинхронная операция
выполниться после того как данный поток будет закрыт, а продолжение работы
сопрограммы начнется именно с того момента где она была остановлена в
предыдущий раз.
26
По сути, решая использовать сопрограммы можно написать асинхронный
код в синхронном стиле (рис. 4). То есть появляется использовать преимущества
обоих подходов, избегая части их недостатков. Нельзя сказать, что это
абсолютное решение, однако, для некоторых задач такой подход будет
наилучшим решением.
Концепция сопрограмм известна давно. Однако споры о введении их в
стандарт языка c++ пока не завершились успехом. Этот вопрос пока отложен до
грядущего стандарта. Гор Нишанов (входит в комитет стандартизации с++)
активно агитировал за их внедрение в c++. В своих докладах он приводит такие
результаты тестирования нескольких реализованных им программ, как с
использованием сопрограмм, так и без них (таблица 2). Этот пример позволяет
оценить возможную пользу сопрограмм при их применении.
Таблица 2 – Результаты тестирования
Простая
реализация
Оптимизация
кода
Устранение
аллокаций
Скорость обработки данных,
Mb/s
(без
(сопрограммы)
сопрограмм
380
495
Размер исполняемого файла, Kb
(без
сопрограмм)
30
(сопрограммы)
485
1028
30
25
690
1028
28
25
25
В таблице 2 простая реализация программы принимающей по TCP –
написанная на основе примера к библиотеке boost.asio, реализованной на основе
машины состояний (рис. 5). Версия «Оптимизация кода» – оптимизация кода
первой программы с использованием возможностей более поздних стандартов
языка с++ (до с++14). «Устранение аллокаций» – устранение аллокаций для кода
без сопрограмм. Реализация сопрограмм не изменилась.
27
Рисунок 5 – Машина состояний
Исходя из таблицы 2 можно сделать вывод что использование сопрограмм
может приносить пользу, но нельзя сделать однозначный вывод об этом, так как
для некоторых задач результаты могут быть обратными. Поэтому необходимо
использовать сопрограммы лишь в тех случаях, когда это может принести
пользу.
Использование
сопрограмм
может
положительно
сказаться
на
эффективности работы разрабатываемого сервиса. Однако окончательное
решение об их полезности возможно только после реализации программы.
Заранее можно лишь предположить, что данное решение может помочь с той
или иной задачей.
1.2.5 Межпротокольное взаимодействие
Принцип работы разрабатываемого сервиса будет описан в следующих
главах
данной
работы,
межпротокольного
однако
взаимодействия.
необходимо
Данные
сказать
собранные
о
принципах
получаемые
от
контролируемых объектов формируются по специальным протоколам, которые
зависят от установленного оборудования. Их существует большое количество.
Поэтому необходима добавлять или изменять реализованные в телематической
системе протоколы.
Под протоколом обычно понимается набор правил и
соглашений, используемых при передаче данных [19]. Данное определение
варьируется в зависимости от места применения.
28
Помимо протоколов обеспечивающего прием данных и отправку данных
на устройство сбора телеметрии необходимой частью является протоколы для
отправки данных от приемной части основному серверу. Для подобной цели
планируется использование как основного протокола протокол AMQP. Однако
важной особенностью разрабатываемого сервиса будет то, что будет реализована
возможность добавления других протоколов. AMQP (Advanced Message Queuing
Protocol) — открытый протокол для передачи сообщений между компонентами
системы. Основная идея состоит в том, что отдельные подсистемы (или
независимые
приложения)
могут
обмениваться
произвольным
образом
сообщениями через AMQP-брокер, который осуществляет маршрутизацию. Этот
протокол гарантирует доставку, распределение потоков данных, подписку на
нужные типы сообщений [20].
Существует большое количество протоколов, которые можно исопльзовать
для
передачи
данных
телеметрии.
Обычно,
каждый
производитель
телематического оборудования разрабатывает собственный протокол. При этом
протоколы могут отличаться очень сильно друг от друга. Начиная от установки
соединения с объектом приема данных (разрабатываемый в данной работе
сервис), заканчивая форматом представления и шифрованием поступающих
данных. Несколько таких протоколов будет использовано в данной работе.
Помимо передачи данных внутри платформы, может потребоваться
отправка данных куда-нибудь еще, например сторонняя система обработки
данных, если это необходимо. Разрабатываемый сервис должен давать
возможность менять и настраивать необходимые способы пересылки данных. То
есть необходимо, чтобы изменение части параметров передачи данных, таких
настройка протоколов, портов, и т.п. происходило без изменения основного кода
сервиса. Для этого различные настройки могут выноситься в конфигурационный
файл, и для чтения новых настроек из него достаточно будет изменить файл и
перезапустить сервис
29
2. Исследование области применения сервиса межпротокольного
взаимодействия
2.1 Предметная область
В
настоящее
время
тематические
системы
развиваются
очень
стремительно и спрос на подобные системы огромен. Однако необходимо
помнить, что в процесс разработки таких систем необходимо учитывать большое
количество различных особенностей. Разрабатываемая система, не может
работать отдельно. Для выполнения своих функций она должна быть частью
одной или нескольких телематических систем.
Телематические системы имеют очень сложную структуру, состоящую из
множества незаменимых частей. Незаменимым элементом подобных систем
является
подсистема,
связанная
с
организацией
межпротокольного
взаимодействия. Подобные сервисы должны обеспечивать первичную обработку
данных и реализацию правильного и безопасного взаимодействия в процессе
обмена данными в телематической системе. Сложность подобных систем может
разниться в зависимости от определенных требований к конкретной системе.
Может
меняться
как
величина
нагрузки
(количество
единовременных
соединений между участниками обмена данными), так и степень обработки
данных (вплоть до полного разбора данных от объектов контроля и
регулированием процесса отправки команд на них). Контроль нагрузки
осуществляется
путем
запуска
дополнительных
экземпляров
сервиса
межпротокольного взаимодействия. Данный процесс не входит в данную работу,
но разработанный сервис предоставляет возможность параллельной работы
нескольких инстансов с программой.
Адаптивность системы межпротокольного взаимодействия является одной
из ключевых особенностей. Под адаптивностью здесь понимается как
возможность работы с различными протоколами обмена данных, так и настройка
процесса формирования данных.
30
2.1.1 Транспортная телематика
Появление
относительно
дешевых GPS-датчиков
и
стремительное
распространение систем сотовой и спутниковой связи привело к созданию самых
разнообразных
компьютеризированных
систем
управления
наземным
транспортом, позволяющим диспетчерам в реальном времени не только
отслеживать на экранах мониторов истинное местонахождение интересующих
их транспортных средств (легковых и грузовых машин, самосвалов, автобусов и
т. д.), но и следить за текущим состоянием этих машин (уровнем топлива в
бензобаке,
температурой
в
салоне
автобуса
или
в
грузовом
отсеке
рефрижератора и т. д.). При этом серверы, на которые стекаются данные от
бортовых систем, могут находиться как в самих диспетчерских, так и в облаках
(частных или публичных).
Особенностью данной отрасли является то, что на рынке телематических
устройств существует большое количество производителей-конкурентов. При
этом большая их часть используют собственные протоколы передачи данных.
Иногда они отличаются в деталях, иногда разрабатываются собственные, или
добавляется возможность выбора необходимого решения. Это приводит к тому,
что помимо технических решений, производители разрабатывают также и
собственные телематические решения (телематические системы). Для части
клиентов это не является проблемой, скорее наоборот. Однако такие решения,
добавляют проблем тем кто, например, использует оборудование разных
производителей, или транспортная телематика является лишь малой частью
потребностей.
Интернет вещей также обладает рядом проблем, которые мешают
быстрому развитию отрасли и требуют более сложных решений.
Поэтому
разработанный
расширяемый
сервис
межпротокольного
взаимодействия может занять свою нишу в области Интернета вещей и
транспортной телематики в частности.
31
2.1.2 Проблемы интернета вещей
Интернет вещей позволяет организовать процесс сбора и аналитика
данных
с
огромного
направленности.
числа
различных
Немаловажным
является
объектов
соверешенно
предоставление
разной
возможности
управлять этими объектами удаленно, вручную или с помощью заранее
заданных сценариев, при изменения условий среды и наступлении определенных
событий. Но кроме того устройства могут взаимодействовать друг с другом, без
участия в этом процессе человека. Это позволяет сделать вывод об
перспективности данной концепции. Одним из основных ее преимуществ
является то, что использование данной методологии практически безгранично.
Однако это приводит к большому число проблем, тормозящих повсеместное
развитие и использование IoT.
Большую сложность представляет собой соединение всех цифровых точек
воедино. Особенно трудно разрабатывать и строить системы, которые будут
работать в реальном мире с максимальной пользой. Помимо социальных и
психологических ограничений существуют и технические.[21, c.48]. Одной из
главных проблем современного Интернета вещей является всестороннее
взаимодействие всех участников данной отрасли. В физическом плане для
Интернета вещей проблем и препятствий практически нет. Существует
достаточное разнообразие устройств (датчиков, контроллеров и т.д.), которые
могут быть использованы для работы Интернета вещей. Кроме того, рынок
устройств для IoT постоянно пополняется новыми идеями и технологиями. Это
порождает новые области применения и дает возможность улучшить ситуацию
во внедренных технологиях в Интернете вещей ранее.
Многие IoT решения достаточно узки и специфичны. Они не реализуют
весь возможный потенциал. Интернет вещей подразумевает работу с огромным
количеством данных и устройств. Развитие рынка IoT систем является одной из
косвенных причин того, что процесс обработки большого объема данных в
32
последнее время является перспективным разделом IT (Big Data). Кроме того,
из-за достаточно легкого и недорого доступа к оборудованию у каждого
человека есть возможность попробовать сделать свой “маленький” Интернет
вещей. А с повсеместным внедрением облачных технологий в нашу жизнь это
становится реальным. Небольшое число компаний могут предоставлять услуги
по использованию Интернета вещей в полном объеме. Конечному пользователю
предпочтительней получать готовый продукт, удовлетворяющий всем его
запросам. Так же, важным частью таких продуктов должна быть возможность
его модернизации, как в техническом плане (замена, оборудования, каналов
передачи данных), так и в логическом (изменения модели поведения системы и
т.п.). Полный стек технологий необходимый для создания настолько сложных
систем огромен. Необходимы специалисты разного профиля и уровней
подготовки. Поэтому на данном рынке важно тесное взаимодействие с
компаниями, которые могут дополнять друг друга. В настоящее время многие
крупные компании работают в этом направлении (Microsoft, Amazon), кроме них
существует достаточное число менее известных компаний, которые стараются не
отставать
и
иногда
предлагающих
интересные
решения.
Интересное
особенностью данного рынка является то, что те компании, которые реализуют
полных стек технологий, нередко продают свои продукты по частям.
Однако нельзя сказать, что Интернет вещей это развитая отросль. Он
развивается. Это приводит к возникновению определенных трудностей. Это
новая отрасль, находящаяся в процессе становления. Несмотря на это,
требования к системам связанными с IoT очень высоки. Поэтому участники
Интернета
вещей
постоянно
сталкиваются
с
различными
проблемами.
Сложность систем связанных с Интернетом вещей связана с тем, что задачи,
которые необходимо решить, требуют практически безотказности систем и
быстрого реагирования на внешние изменения. Надежность и скорость работы –
критически важны.
33
Одной из основных проблем современного IoT является разнообразие
стандартов и протоколов внутри него. Это мешает его к простому и
эффективному развитию. С самого появления «умных» устройств и систем, их
создатели подстраивались под свои задачи и задумки. Поэтому появилось
существенно отдаления потребительского Интернета вещей от промышленного.
Разное видение реализации приводит к созданию различных, непохожих друг на
друга,
протоколов.
Поэтому
на
рынке
IoT,
не
последним
фактором
конкурентоспособности и успеха системы является её «многопротокольность».
И для новых компаний это является серьезным препятствием.
Сейчас разобраться с этим чуть проще, так есть стандарты, которые
рекомендуются для решения конкретных задач Интернета вещей. Но единой
стандартизации все равно нет. Все сводится к рекомендациям. Это иногда
приводит к тому, что используя один и тот же протокол передачи данных, нет
гарантий, что данные будут приходить в одинаковом формате. Для примера
возьмем условную температуру, которую передают некие устройства. Эти
сведения могут быть переданы в разном формате. Для примера информации о
том, что температура равна 26 градусов по Цельсию. Передачи числа 26 может
меняться в зависимости от реализации. Например, число преобразованное в
текст (символы «2» и «6»); в дробном (26.0), если устройство поддерживает
более точное измерение; в шестнадцатеричной системе счисления, то есть если
пытаться получить текст, то получим символ ascii «&»; или значение будет
передано вместе с другими полями, внутри какого-нибудь текстового формата
обмена данными(например, json) и т.д.. Такие ситуация возникаю постоянно. И с
этой проблемой стандартизация помочь пока не в состоянии. Поэтому
современные системы обработки данных для Интернета вещей должны быть
максимально гибкими для возможной обработки различного преобразования
данных.
Некоторые организации занимающиеся проблемами стандартизации
выделяют несколько основных протоколов передачи информации, при этом в
34
зависимости от их сильных и слабых сторон указывают рекомендуемые области
применения. Так для промышленного Интернета вещей основным документомрекомендацией
является
издание
подготовленное
the
Industrial
Internet
Consortium Connectivity Task Group, в состав которого входят люди напрямую
свзанные с этой частью Интернета вещей (представители компаний Lead, RTI,
Cisco, Ericsson, Nokia и многих других). На 2017 год вышло уже пять редакций и
последняя называется «The Industrial Internet of Things Volume G5: Connectivity
Framework»[22], а учитывая скорость развития Интернета вещей, эта редакция
далеко не последняя. В этом издании можно почерпнуть неполное, но довольно
подробной описание технологий промышленного Интернета вещей, с их
классификацией и определением их особенностей. В вкратце разнообразие и
структуру промышленного интернета вещей представлено на рис. 6 [22, с.46].
Рисунок 6 – Стандарты передачи данных IoT
На данной схеме видно, что существует определенное разнообразие даже
внутри промышленного интернета вещей, причем оно проявляется почти на всех
уровнях передачи данных. При этом нельзя забывать, что различия могут быть и
внутри одного протокола.
35
Отсутствие единых стандартов и протоколов замедляет развитие области
Интернета вещей. Это проявляется как при добавлении новых технологий и
устройств внутри одной области применения, так и при пересечении нескольких
отраслей. При этом затрачивается дополнительные человеческие и временные
ресурсы и теряется экономический и практический потенциал Интернета вещей.
Придумываются собственные узконаправленные решения, которые мешают
(замедляют) повсеместному развитию всего Интернета вещей.
Следующая причина замедления развития интернета вещей является
безопасность. Сейчас сложно придумать область жизни, где невозможно
внедрение Интернета вещей, поэтому возможность перехвата управления
умными вещами может пугать. Начиная от, казалось бы, безобидных шалостей
предоставления доступа к лампочке в чужом доме до нарушения в работе целого
предприятия, что способно привести к катастрофе. Поэтому вместе с развитием
самого Интернета вещей обязательно параллельно должна вестись работа по его
защите. Иначе, развитие технических возможностей Интернета вещей станет
бессмысленным. Угрозы для Интернета вещей разнообразны. К ним относят все
угрозы присущие информации: угрозы конфиденциальности, целостности,
доступности. Кроме того особенно выделяется угроза контролю доступа к тем
элементам системы, которые находятся под управлением Интернета вещей.
Должен быть обеспечен полный контроля доступа ко всем техническим
элементам системы. Для этого необходимо обеспечить надежные механизмы
аутентификации и авторизации пользователей. А с учетом развития M2M
(Межмашинное взаимодействие, машинно-машинное взаимодействие, Machineto-Machine) телематики так же обеспечить должный уровень аутентификации на
этом уровне. Эта тема для отдельного исследования, однако стоит отметить что
данной проблематике стоит уделить особое внимание при реализации IoT
систем.
Сейчас существует большое количество методов контроля доступа к
оборудованию и информации на разных уровнях таких систем. Так, шифрования
36
данных на уровне протоколов передачи данных является чуть ли не
обязательным условием безопасного Интернета вещей.
С безопасностью так же связана и другая проблема: боязнь людей
(конечный пользователей Интернета вещей) как новых технологий, так и
сопутствующих им угроз безопасности как данных людей, так и их здоровья. Но
это проблема скорее поверхностная, и будет решаться за счет популяризации
Интернета вещей и некоторого времени на адаптацию людей к новому
окружению. Но в текущем прогрессивном мире это проблематика кажется уже
не слишком серьезной.
Говоря о технических сложностях Интернета вещей, важной становятся не
сами технические решения по управлению устройствами и сбора их данных, а
проблема передачи данных. Надежные проводные соединения не удобны в
некоторых случаях (например, умные машины), а беспроводная связь зачастую
не обеспечивает необходимой зоны покрытия, да и в целом, менее надежная.
Кроме того, затраты на связь могут быть очень дороги. Часто требуется
предоставление контроля над оборудованием в режиме 24/7.
Помимо мониторинга и сбора данных IoT системы должны реагировать на
изменения, происходящие как внутри системы, так и вне нее. Поэтому в
серьезных системах просто необходима возможность автоматизированного
контроля состояния, с минимальным участием этом деле человека. То есть
простые оповещения пользователя о критических изменениях ситуации на его
объекте — это лишь минимальная часть таких систем. Необходимо обеспечить
возможность настройки поведения системы при изменениях каких-либо
критериев. Это подводит к одной из основных проблем IoT. IoT — это сложно.
Большое количество технических аспектов приводит к необходимости долгой и
детальной разработке, где ошибки могут привести к серьезным проблемам.
Нужно учитывать множество факторов, опыт других разработчиков. Кроме того,
в некоторых случаях важна быстрота реакции системы на малейшее изменения.
37
Это приводит к тому, что на этапе проектирование IoT систем необходимо
четко осознавать все детали поставленных задач, знать основные технологии
реализации таких систем, чтобы выбрать наиболее эффективный способ
реализации и выбор оборудования, чтобы добиться необходимого уровня
стабильной работы системы.
2.2 Поиск и изучение готовых решений в области применения сервиса
При поиске аналогов разрабатываемой системе были изучены как
узкоспециализированные
платформы
IoT,
так
платформы
с
большими
возможностями. В этих платформах по разному обеспечиваются обработка
данных при межпротокольном взаимодействии. Иногда эти вообще не имеют
такого
взаимодействия.
взаимодействия.
В
этом
Большинство
случае
подобных
усложняется
систем
процесс
M2M
специализируются
на
отдельных протоколах или определенных видах оборудования, понемногу
расширяя
список
поддерживаемого
оборудования
на
основе
текущих
потребностей.
2.2.1 Интернет вещей
Сейчас рынок телематических систем для Интернета вещей достаточно
широк. Интернет вещей – динамически развивающаяся область. Даже крупные
игроки рынка телематических услуг зачастую не сделали готовых решений для
всех разделов данной отрасли. Однако, из описанных выше проблем Интернета
вещей,
подобные
платформы
часто
специализируются
на
нескольких
протоколах.
Для таких протоколов как, например, MQTT и COAP, (описанных в
предыдущем разделе), существуют несколько видов решений. Как полноценных
платформ, с широкой возможностью настройки, так и небольших, самописных
38
библиотек распространяемых на ресурсах подобных www.github.com. Все эти
решения имеют разные достоинства и недостатки.
Например, для MQTT одной из самых известных платформ является
Mosquitto.
Mosquitto
–
платформа,
реализующая
возможность
обмена
сообщениями между компонентами системы на базе стандартов протокола
MQTT версии 3.1 и 3.1.1. Mosquitto написан на языке C и распространяется под
лицензией BSD.
К проектам, использующим данную платформу-брокер
относятся, например, Facebook Messenger и Gaian DB. Данная платформа
представляет собой инструмент как для получения данных отправленных с
помощью протокола MQTT, так и генерации их. Работает как сторонний сервер,
расположенный в сети интернет. К недостатку данной платформы относится
узкая направленность, и проблемы с детальной настройкой и вариативностью, но
для небольших по требованиям проектов она подходит хорошо. Исопльзование
этой платформы можно внедрить и в разрабатываемый сервис, однако это
повлияет на скорость работы, так как потребуется дополнительное время на
взаимодействие с этой платформой.
Из больших компаний, глубоко углубившихся в Интернет вещей, следует
отметить
Microsoft
и
Amazon.
Microsoft
предоставляет
возможность
использовать их облако (Microsoft Azure) для различных целей. Это избавляет от
необходимости уделять внимание техническому обеспечению. Azure IoT
обладает становится хорошей базой для создания собственных решений
связанных с IoT. Но обладает более сложным процессом интеграции с менее
популярными протоколами.
Как и Azure IoT, Amazon IoT предоставляет хорошее решение не
зависящее от аппаратной составляющей, однако существует ряд ограничений
(например выбор протокола), не позволяющей применять его повсеместно.
Выбор решения напрямую зависит от поставленных задач и бюджета
проекта. Однако в современных реалиях, когда в Интернет вещей существует
39
достаточно большой выбор IoT платформ, они должны предоставлять какие-то
конкурентные преимущества, что бы выбор падал именно на них.
2.2 Транспортная телематика
Транспортную телематику можно отнести к Интернету вещей. Рассмотрим
ёё более детально на нескольких примерах. Телематические технологии
представлены
на
автомобильном
рынке
многими
производителями,
но
телематический сервис в настоящее время еще развивается, хотя конкуренция на
данном рынке просто огромна. Среди доступных телематических систем ярким
примером является система OnStar (www.onstar.com) от General Motors,
предлагаемая как стандартный элемент автомобилей и грузовиков, продаваемых
в Северной Америке. OnStar предлагает средства повышения безопасности,
сервис удаленной диагностики автомобиля, отчет о которой направляется
пользователю по электронной почте, систему навигации Turn-by-Turn — в виде
подробных голосовых инструкций следования по маршруту и выполнению
поворотов, необходимых для достижения пункта назначения. Компания GM
также осуществила возможность коммуникации между двумя автомобилями. На
основе платформы WMfA разработана система Blue&Me Fiat, поддерживающая
hands-free звонки, подключение сотовых телефонов и медиаплееров через
Bluetooth и USB интерфейсы, а также систему навигации, которую можно
активировать голосом [23].
1
Помимо гигантов авторынка телематический услуги предоставляют и
компании,
для
которых
это
является
основным
источником
дохода.
Потребителями телематических систем могут быть разные организации, в том
числе государственные и муниципальные структуры. Например, мониторинг
автобусов, троллейбусов, трамваев сейчас работает во всех крупных городах
нашей страны, при этом зачастую подкреплен дополнительным функционалом.
40
Согласно исследованию компании "Омникомм-Сервис"[23] российского
рынка транспортной телематики, рынок вырос в количественном выражении на
340 тыс. оборудованных транспортных средств (ТС). Темпы роста, в сравнении с
2013 годом, отрицательные.
Количество подключений снизилось на 15% – исследование 2013 года
показывало нулевой темп роста 2013 года по сравнению с 2012 годом. Менее
заметно спад сказался на количестве установленных датчиков уровня топлива –
5% роста – 210 тыс.установок в течение 2014 года.
Крупнейшими игроками на рынке навигации в России являются компания
"Навигационно-информационные
федерального
оператора
в
системы"
области
(НИС),
спутниковой
получившая
навигации,
и
статус
«М2М
Телематика».
Другим заметным игроком является компания Русские навигационные
технологии. Компания, по собственным оценкам на 2014 год, занимает 24%
российского рынка систем мониторинга автотранспорта, входя вместе с «М2М
Телематика» и «Навигатор Технолоджи» (разработчик «Единой Национальной
Диспетчерской Системы») в «большую тройку» поставщиков.
Детально разберем схему работы телематической системы на примере
компании
«СпейсТим».
специализированное
На
бортовое
транспортное
ГЛОНАСС/GPS
средство
устанавливается
оборудование,
которое
принимает и обрабатывает информацию о географических координатах,
полученных с навигационных систем ГЛОНАСС и GPS, а так же данные с
периферийных устройств (датчики уровня топлива, камеры, тревожная кнопка,
исполнительные устройства и др.).
После того, как все данные получены и обработаны оборудованием,
информация в режиме реального времени независимо от времени суток и
погодных
условий
передается
по
каналам
беспроводной
связи на
телематический сервер. Телематический сервер – основной инструмент для
сбора,
обработки,
промежуточного
хранения,
персылки
данных
для
41
последующей
обработки
и
анализа данных,
поступающих
с
объектов
мониторинга. Возможна реализация системы как на базе собственного
телематического сервера предприятия, так и на базе сервера оператора.
Далее
данные
с
сервера
через
сеть
Интернет
поступают
на
автоматизированное рабочее место (АРМ) диспетчера. С помощью специального
программного обеспечения (ПО) диспетчер оперативно в режиме реального
времени управляет процессами мониторинга, контролирует работу водителей,
получает тревожные сообщения в случае возникновения внештатных ситуаций и
принимает регламентированные меры, контролирует технические параметры
объектов,
ведет
статистику
и
учет
работы,
анализирует
ситуацию.
Автоматизируются рабочие места диспетчеров, логистов, IT–специалистов,
работников служб эксплуатации, директоров по безопасности и руководителей
предприятий в зависимости от бизнес-потребностей и политики безопасности
предприятия.
В
состав
аппаратно-программных
комплексов систем
мониторинга
транспорта входят:
– телематический сервер на базе телематических платформ;
– клиентское и серверное ПО автоматизированных рабочих мест
пользователей;
– картографическое (ГИС) программное обеспечение;
– навигационно-связное
ГЛОНАСС
оборудование
контролируемых
объектов.
В системе
мониторинга
транспорта
используются
инновационные
технологии приема и передачи данных:
– cпутники глобальных систем позиционирования ГЛОНАСС и GPS
отправляют сигналы, позволяющие вычислять и обрабатывать информацию о
географических координатах объекта;
42
– сети мобильной связи GSM / GPRS обеспечивают обмен данными между
ГЛОНАСС
оборудованием
контролируемых
объектов и
телематическим
сервером;
– всемирная
сеть
Интернет
обеспечивает
обмен
данными
между
телематическим сервером, компьютером пользователя и системами связи.
Конечно, здесь были описаны далеко не все разработки в области
телематических систем. Сейчас многие небольшие компании пытаются занять
эту нишу, и спрос на их услуги достаточно высок.
2.3 Выводы об области применения
По итогу исследований области применения, можно сделать вывод, что
при необходимости работы с разными типами оборудования и протоколами
передачи данных сильно ограничен выбор решений. И для некоторых задач,
придется использовать разные платформы, возможно специально дорабатывая
их своими надстройками или создавать собственные решения. Это приводит как
к усложнению решений, так и банальному неудобству, даже если поставленные
цели будут достигнуты.
Расширяемый сервис межпротокльного взаимодействия будет интересен в
первую очередь новым компаниям, которые хотят занять определенное место на
рынке IoT решений. Использование сервиса межпротокольного взаимодействия
предоставляет возможность абстрагироваться от различий контролируемый
устройств. Сервис межпротокольного взаимодействия предоставляет инструмент
по быстрому и удобному внедрению нового оборудования
в работу
телематической системы. Разрабатываемый сервис предоставляет широкую
возможность по быстрому расширению области применения любой IoT
платформы. Да, для некоторых платформ будет достаточно узких решений
(иногда даже нескольких). Однако использование именно этого сервиса
позволяет избежать проблем в случае желания увеличить область покрытия IoT.
43
3 Проектирование расширяемого сетевого сервиса межпротокольного
взаимодействия для высоконагруженных телематических систем с
асинхронной сервис-ориентированной архитектурой
3.1 Задачи поставленные перед разработкой сервиса
Для
телематических
взаимодействия
с
платформ
устройствами
важной
контроля.
частью
является
Необходимо
способ
единообразия
получаемых данных от устройств вне зависимости от протокола передачи
данных, который они используют. Данная работа описывает разработку
расширяемого
сетевого
высоконагруженных
ориентированной
устройства,
сервиса
межпротокольного
телематических
архитектурой.
использующие
систем
Данный
различные
с
сервис
взаимодействия
асинхронной
позволяет
протоколы
как
для
сервис-
обрабатывать
транспортного
уровня(TCP, UDP, TLS и др.), так и протоколы определяющие обмен данными с
сервером.
Для разработки данного сервиса важными являются следующие задачи:
– возможность первичной обработка данных;
– взаимодействие с объектами контроля;
– безопасность данных;
– оповещение о состоянии объектов;
– возможность работы в высоконагруженных телематических системах;
– возможность внедрения новых протоколов устройств и передачи
данных.
Сервис
должен
запускаться
в
фоновом
режиме
(как
демон),
Контролировать подключения контролируемых объектов и подключения к
сторонним сервисам.
Для проектируемого сервиса было выбрано название RicGate. Данной
название состоит из двух слов. Ric – сокращение от Rightech IoT Cloud (см.
44
далее) и Gate (англ.) – ворота, это слово достаточно определяет назначение
сервиса.
3.2 Применение разработанного сервиса в платформе для Интернета
вещей Rightech IoT Cloud
Рисунок 7 – Платформа Интернета Вещей Rightech IOT Cloud
Примером использования сервиса межпротокольного взаимодействия
может являться облачная платформа Rightech IOT Cloud (RIC). На рис. 7
представлена
её
упрощенная
схема.
Данная
платформа
предоставляет
возможность подключать различные устройства, относящиеся к Интернету
вещей, получать с них данные и управлять этими объектами. Платформа
абстрагируется от конкретных устройств, воспринимая как однотипные объекты
для каждого из которых настроен свой способ взаимодействия с ним (формат
передаваемых данных, команды и т.д.).
На данной схеме RicGate – расширяемый сетевой сервис, через который и
происходит непосредственное взаимодействие платформы RIC с объектами
45
контроля. RicGate – это промежуточное звено между основной частью
платформы и контролируемыми устройствами. С помощью платформы RIC
обеспечить сбор и обработку данных, а так же управление объектами контроля,
как в ручном режиме, так и путем создания сценариев поведения. Кроме того,
если это предусмотрено платформой, возможно взаимодействие между
устройствами, без участия основной платформы: устройство1 – RicGate –
устройство2. Это немаловажная особенность сервиса RicGate. Логика работы с
устройствами контроля и обработка их данных задается в входных плагинах.
Выходные плагины предназначены для тех же функций, но уже для сторонних
сервисов. Для добавление нового вида оборудования, предоставляется базовый
класс, на основе которого пишется плагин, аналогичная ситуация с добавлением
нового адресата получения данных от объектов контроля.
3.3 Разработка расширяемого сетевого сервиса межпротокольного
взаимодействия
Разработанный сервис не предусматривает взаимодействие с человеком,
поэтому для описания внутренних процессов была и использована диаграмма
вариантов использования (рис. 8), где актёрами являются сущности которые
взаимодействуют с системой, т.е. контролируемые объекты и сторонние сервисы
целостной телематической системы. RicGate не может работать без сторонних
сервисов, отвечающих за процесс управления объектами, и являющимся
получателем данных от объектов контроля. Частично обработку данных можно
возложить на RicGate. Для этого был разработан механизм, который позволяет
определить формат передаваемых данных – строка, число или булевый тип.
Обычно фрагмента данных образует связку ключ-значение. Поэтому несложно
организовать
возможность
при
обработке
авторизации
возвращать
дополнительную информацию по обработке конкретных полей по ключу. После
разбора данных полученных от устройства происходит их обработка в
46
соответствии с информацией, формируемой на процессе авторизации. Данные
передаваемые при авторизации заданы моделью объекта в платформе. Это
необходимо чтобы была возможность привести данные отправляемые из RicGate
к единому формату.
Для описания концептуального представления системы была использована
диаграмма вариантов использования (рис. 8).
Рисунок 8 – Диаграмма вариантов использования
На данном рисунке представлены процессы авторизации, отправки данных
и получения команд. Любое устройство, которое пытается подключиться к
платформе должно пройти процедуру авторизации. То есть происходит проверка
идентификатора устройства, логина и пароля для доступа к устройству. Данный
процесс может привести как к разрыву соединения с устройством, так и к
успешной авторизации и предоставление возможности дальнейшей работы
устройства в системе. В зависимости от протокола передачи данных для
контролируемых устройств разрыв соединения может сопровождать отсылкой на
устройство данных о причине разрыва соединения.
47
Процесс отправки команд сопровождается контролем над временем ответа
устройства и в зависимости от настроек отправкой повторной команды или
разрывом соединения.
Для того что бы описать динамические аспекты процесса обработки
одного объекта (поведение сервиса) была использована диаграмма состояний
(рис. 9).
Рисунок 9 – Диаграмма состояний
Диаграмма состояний (рис. 9) позволяет описать реакцию сервиса на
восприятие конкретных событий, связанных с объектами контроля или
командами от полученных от других сервисов. После запуска RicGate ожидает
подключения по портам указанным в настройках каждого входного плагина, то
есть существует связка протокол-порт для ожидания подключений объектов
контроля. Если необходимо настроить несколько протоколов на один порт, то
48
данную проблему можно попробовать решить внутри конкретного плагина. Это
добавит трудности в реализации плагина, но вполне выполнимо. Больше
протоколов, которые используются в реализации передачи данных в IoT или
подобных концепциях после установки соединения отправляют данные для
авторизации. Набор этих данных может быть разный, но обязательно должен
присутствовать идентификатор устройства (например, IMEI) и почти всегда
связка логин-пароль для авторизации устройства.
Авторизация устройства, если она необходима, происходит в другом
сервисе телематичекой платформы. В данный момент запрос авторизации
формируется
в
формате
протокола
JSON-RPC
и
перенаправляется
в
соответствующие выходные плагины для отправки, но возможно использование
разных способов передачи данных, например, отправка данных в другую, не
связанную с телематической платформой, систему. Примером такой системы
может быть пересылка данных полученных от автомобилей, на сервер
Управление Государственного Автодорожного Надзора (УГАДН) по протоколу
EGTS. Все зависит от реализации выходного плагина.
После успешной авторизации, работа сервиса с объектом происходит по
следующим возможным путям:
– получение
и
отправка
обработанных
данных
от
объекта
в
соответствующий выходной плагин;
– подписка на какой-либо канал, с целью получения данных напрямую
через RicGate с других устройств;
– получение команды через выходной плагин и отправки команды на
объект контроля.
Реализация
расширяемого
сетевого
сервиса
межпротокольного
взаимодействия для высоконагруженных телематических систем с асинхронной
сервис-ориентированной архитектурой сложна для представления и обладает
множеством особенностей, которые усложнят понимания принципов реализации
системы на диаграмме классов. Поэтому на диаграмме классов (рис.10)
49
представлены элементы позволяющие отразить основную идею реализации
сервиса.
Рисунок 10 – Диаграмма классов
Подключение – это класс, который управляет соединениями с объектами
контроля. Один экземпляр класса – это одно подключение. Данный класс может
работать с разными протоколами транспортного уровня стека TCP/IP (TCP, UDP,
TLS). В будущем планируется добавить поддержку DTLS. Это важно так как,
один и тот же протокол передачи данный устройства контроля, в зависимости от
настроек конкретных устройств, может использоваться для разных протоколов
транспортного уровня (например, TCP и TLS).
Класс
экземплярами
КонтрольПодключений
классов
обеспечивает
ВходнойПлагин
и
маршрутизацию
ВыходнойПлагин.
между
Классы
ВходнойПлагин и ВыходнойПлагин – базовые классы для создания уникальных
дочерних классов для взаимодействия с объектами контроля и сторонними
сервисами соответственно. Эти классы предоставляют функции доступные из
ядра системы, для управления ими. Их описание будет дано в подразделе 4.3.
50
3.4 Архитектура сервиса
Рисунок 11 – Представление архитектуры сервиса с помощью диаграммы
компонентов
Разрабатываемый сервис должен обеспечивать взаимодействие объектов
контроля и других сервисов телематической платформы. На рис. 11 видно,
взаимодействие частей системы происходит через ядро. Функция ядра –
запустить
работу
элементов
системы.
Сначала
ядро
подгружает
все
необходимые плагины из соответствующей директории. Затем запускаются
механизмы контроля подключений для объектов по протоколам TCP, UDP, TLS
(при необходимости данный список может быть расширен). Одновременно с
этим, происходит подключение сервиса к сторонним сервисам. В практическом
использовании применяется платформа RabbitMQ, которая широко используется
для
построения
сервис-ориентированной
архитектуры,
и
Redis.
При
необходимости разработанный сервис можно подключить и к другим сервисам,
написав самостоятельный плагин наследуясь от соответствующего класса.
Описание реализации входных и выходных плагинов и описание их работы
51
будет дано в разделе 4.3. Аналогично загружаются входные плагины, с той лишь
разницей, что используется свой базовый класс, отмеченный на диаграмме
классов (рис. 10) на схеме как ВходнойПлагин.
В контроле подключений из конфигурационного файла считываются
данные о маршрутах отправки данных, т.е. маршрут отправки данных от
объектов контроля до сервисов, которые являются получателями данных.
Полученную архитектуру можно условно поделить на три основные части.
Первая – подсистема обработки данных от контролируемых объектов и
переводу данных(например, команд) от телематической платформы в команды
поддерживаемые протоколом передачи данных контролируемых объектов.
Вторая – подсистема взаимодействия с другими сервисами телематической
системы (передача данных от полученных от объекта контроля другим сервисам
и прием команд от них для передачи объекту контроля).
Третья – ядро системы обрабатывающей подключения по протоколам
транспортного уровня и обеспечивающая взаимодействие двух других частей
системы и управление внутренними процессами.
3.4.1 Ядро системы
В ядре системы происходит контроль подключений. Сюда входит
управление существующими подключениями и обработка новых подключений.
Для наилучшей работы ядро системы должно иметь асинхронную архитектуру,
иначе для работы с большим числом подключений придется использовать, либо
несколько копий сервиса (для разных протоколов своя копия), либо
использовать дополнительное оборудование для распределение нагрузки. Так
как количество устройств, которые могут быть подключены к Интернету вещей
растет постоянно это обязательная часть сервиса.
Загрузка сервиса происходит следующим образом: При старте демона
происходит загрузка всех плагинов из указанных в файле конфигурации
52
директорий. Это необходимо чтобы получить имена всех доступных сервису
плагинов. Затем загружаются необходимые выходные плагины и создаются
необходимые им соединения (например, к другим сервисам платформы).
Создается список маршрутизации для связок «входной плагин – выходной
плагин». Параллельно с загрузкой выходных плагинов создаются необходимые
слушатели для необходимых выходных плагинов (TCP, TLS, UDP и др. если это
возможно). Приняв новое соединение, RicGate создает объект соответствующего
ему класса (определяется по порту подключения).
Ядро обеспечивает взаимодействие как между другими частями сервиса,
так и между объектами контроля и телематической платформой. Сервис
межпротокольного взаимодействия можно назвать воротами, через которые
происходит обмен данными телематической системы и объектами контроля.
Помимо всего этого к функциям ядра относится информирование
выходных плагинов о состоянии объектов контроля. В случае изменения
состояния объекта контроля ядро отправляет об этом информацию в выходные
плагины. Всего возможно три вида сообщений: online, offline, reestablished.
Последнее событие соответствует переподключению объекта с тем же
идентификатором. При этом старое соединение разрывается, и заменяется
новым.
3.4.2 Подсистема обработки данных от объектов
Данная подсистема должна обеспечивать принятых ядром системы
данных, их разбор и первичную обработку. Ядро обрабатывает подключение и
от объекта и определяет протокол, с помощью которого необходимо
обрабатывать эти данные.
Взаимодействие
с
сетью
осуществляется
через
интерфейс
предоставляемый входному плагину, который можно разделить на несколько
категорий:
53
– сетевые запросы (запись и чтение в/из сокета);
– запросы обработки данных (авторизация, отправка телеметрии, отправка
ответа на команду и прочее);
– вспомогательные функции (таймер, выполнение функции в сетевом
соединении, принудительное закрытие соединения и др.).
Разделение протоколов может проходить явно, например, по порту к
которому подключается объект. Затем данные передаются соответствующему
обработчику в сервисе обработки данных. Здесь данные
разбираются,
выбираются необходимые данные, определяется корректность данных и
формируется ответ для объекта контроля (если это необходимо). За контроль
поведения объекта отвечает выходной плагин. При необходимости вызывается
метод удаляющий объект, а так как в ядре храниться лишь ссылка на объект
(используется механизм умных ссылок), то при необходимости происходит
полное удаление всех связок по данному объекту из сервиса. При этом в
зависимости от данных, от их корректности, определяется, как необходимо
дальше
взаимодействовать
с
объектом
контроля,
например,
разорвать
соединение, отправить ответ или отправить какую-либо команду. Помимо
команд необходимых для конкретного протокола/оборудования, были добавлена
возможность обработки двух дополнительных команд:
‒ разорвать соединение с объектом;
‒ обновить модель объекта, хранимую в RicGate.
Первая команда необходима чтобы заставить объект переподключиться к
RicGate разорвав с ним TCP-соединение, в том случае если ошибка в
полученных данных была обнаружена уже после того как RicGate отправил
данные по назначенному маршруту или по каким-то иным причинам. Вторая
команда позволяет заменить модель объекта внутри RicGate на более новую
версию. Модель объекта позволяет определить необходимые данные для
отправки и их формат.
54
Сервис
возможность
межпротокольного
дополнения
числа
взаимодействия
обрабатываемых
должен
поддерживать
протоколов,
поэтому
необходимо оставить возможность как добавлять новые протоколы, так и
изменять старые.
3.4.3 Подсистема взаимодействия с другими сервисами
Данная подсистема обеспечивает процесс передачи данных как от
объектов контроля к другим сервисам телематической системы, так и от
телематической системы к объектам контроля (например команды для них).
Для выполнения данных задач обычно не требуется большого числа
протоколов. Иногда можно обходится одним или двумя, однако тут тоже
необходимо обеспечить возможность модификации и добавления протоколов на
случай изменения отдельных аспектов передачи данных и внедрения новых
технологий.
Кроме того данная подсистема позволяет расширить возможность RicGate.
К уже реализованным возможностям можно отнести пересылку данных от
транспортных средств на сервера Управление Государственного Автодорожного
Надзора (УГАДН) по протоколу EGTS в соответствии с одним из приказов
МинТранса РФ[24]. Это опциональная возможность внутри RicGate является
незаменимой
для
пользователей
телематической
системы,
оборудование
которых является устаревшим и не предусматривает отправку данных на
указанные выше сервера. Кроме того, примером может служить реализация
механизма подписки на каналы внутри протоколов MQTT и COAP. Можно
создать выходной плагин, который будет переправлять данные от одного
оюъекта другому. Это позволяет организовать реализацию механизма подписки
внутри RicGate. Это позволяет обеспечить M2M взаимодействие без участия
телематической платформы. По сути, данный механизм позволяет добавить
55
новые возможности в RicGate без больших трудозатрат, не меняя ядро
программы.
3.5 Применение сопрограмм для реализации асинхронности
Для реализации асинхронного взаимодействия внутри сервиса было
решено использовать такую технологию как сопрограммы.
Сопрограмма (англ. coroutine) — компонент программы, обобщающий
понятие подпрограммы, который дополнительно поддерживает множество
входных точек (а не одну как подпрограмма), остановку и продолжение
выполнения с сохранением определённого положения.
Необходимость использования сопрограмм на практике естественным
образом возникает в случае, когда они связаны с алгоритмами ввода и вывода
[18], что можно отнести и к данной работе, так необходимо реализовать процесс
приема и отправки информации. Основное отличие сопрограмм от синхронной
реализации состоит в том что нет блокирующих операций.
Использование сопрограмм для альтернативы асинхронной реализации не
ново. Их использование, к примеру, можно встретить в таком языке
программирования как Python. Однако в C++ данная абстракция еще не
включена в стандарт языка и неизвестно когда это произойдет (пока инициатива
включить
сопрограммы
в
стандарт
C++17
отклонена
комитетом
по
стандартизации).
Чем интересна абстракция сопрограмм, если ее применять правильным
образом для правильной проблемы, то код не теряет ничего, но при этом может
приобрести в производительности.
Иногда возникает необходимость в разработке алгоритмов сразу для
нескольких компьютерных конфигураций, одни из которых имеют больший
объем памяти, чем другие. В таких случаях можно написать программу в виде
набора сопрограмм и поставить число проходов в зависимость от размера
56
памяти: загрузить сразу столько сопрограмм, сколько поместится, а подачу
данных по недостающим связям обеспечить с помощью подпрограмм ввода или
вывода [18].
3.6 Безопасность
Для Интернета вещей существуют различные угрозы. Важной задачей для
разработанного сервиса является контроль санкционированного доступа к
платформе со стороны устройств. Контроль отправки команд на устройства и
получения доступа к данным осуществляется со стороны основной части
платформы. Необходимо обеспечить проверку подлинности устройств и
контроля над правильностью данных. Ликвидность данных от устройств
контроля обычно осуществляется на уровне протокола передачи данных.
Обычно это различные способы подсчета контрольной суммы пакета данных.
Кроме
того,
многие
протоколы
обеспечивают
передачу
данных
в
зашифрованном виде. То есть можно проверять корректность расшифрованных
данных. Многие протоколы также поддерживают информацию о надежности
данных. Например, для навигационных данных, надежность определяется с
помощью информации о силе сигнала со спутниками.
В последнее время появились случаи различного рода атак на различного
рода системы с помощью Интернета вещей, например за счет эмуляции
большого числа подключений.
В рассматриваемой системе обеспечивается контроль за подключением, то
устройство должно пройти процедуру авторизации. Для разных типов устройств
она может немного отличаться. Например, для некоторых устройств после TCPподключения сервис должен ответить специальным сообщением, и только после
этого устройство отправляет данные авторизации. Но в большинстве случаев
авторизация идет стандартно так, как это представлено на рис. 12 и 13. Только
после авторизации возможно дальнейшее взаимодействие устройства и
57
платформы. В дальнейшем планируется ввести возможность временного
добавления объекта в черный список, чтобы избежать частых попыток
переподключения неавторизованного объекта в систему.
Рисунок 12 – Диаграмма последовательности. Успешная авторизация
Рисунок 13 – Диаграмма последовательности. Отклонение авторизации
Авторизация реализована в наиболее простом и понятном виде. На данном
этапе развития системы было решено не добавлять сервису возможность
самостоятельной проверки объектов. В теории возможно добавить выходной
плагин, который будет, например, сверять данные устройств с данными в базе
данных. Сейчас это реализовано с помощью Rpc-взаимодействия (вызова
58
удаленных процедур (Remote Procedure Call)) с основной платформой. Это было
сделано,
чтобы
уменьшить
нагрузку
на
сервис
межпротокольного
взаимодействия и избежать дополнительных подключений к базе данных.
В большинстве протоколов подходящих для интернета вещей есть
поддержка различных способов сокрытия данных. Некоторые протоколы
являются закрытыми, и без доступа к их описанию набор данных, отправляемых
устройством,
выглядит
как
набор
случайных
байтов.
Это
создает
дополнительные трудности для доступа к данным. Иногда для передачи
большого объема данных данные сжимаются, с помощью специальных
алгоритмов.
Для контроля целостности полученных данных многие протоколы
добавляют к отправляемым полям поле с рассчитанной контрольной суммой
(например, по алгоритму CRC16). При этом большая часть протоколов требует
отправку ответа на их сообщение с результатом обработки данных, т.е.
необходимо отправка информации об успешном получении данных или код
ошибки.
Одним из самых частых способов защиты данных при их передаче от
устройств является использование шифрования. На данный момент для данного
сервиса необходима поддержка использования протоколов транспортного
уровня TLS(Transport Layer Security) и DTLS(Datagram Transport Layer Security).
RicGate поддерживает работу с TLS. Добавление поддержки DTLS планируется
в ближайшем будущем. Управление подключениями закладывается в ядре
RicGate. Эти протоколы являются надстройками над TCP и UDP соответственно.
Для
работы
этих
протоколов
требуется
дополнительно
сертификаты
позволяющие обеспечить безопасный обмен данными. На данный момент два
наиболее часто используемых протоколов Интернета вещей поддерживают эти
технологии (MQTT и COAP соответственно). Однако это не значит что данные
протоколы останутся единственными необходимыми. Так для транспортной
телематики они слабо применимы, хотя и такие решения существуют.
59
4. Реализация сервиса RicGate
4.1 Описание инструментария
Для реализации сервиса межпротокольного взаимодействия был выбран
язык
программирования
C++.
В
данной
работе
решено
использовать
международный стандарт языка C++ (так называемый C++14) «International
Standard Programming Language C++»[25]. В будущем планируется переход на
версию с++17, это обусловлено тем, что стандарт 2017 года не был полностью
утвержден на момент окончания написания выпускной квалификационной
работы.
Помимо средств стандарта языка c++ было решено использовать Boost[26]
– собрание библиотек классов, использующих функциональность языка C++ .
Эти библиотеки позволяют сильно упростить решение многих задач, например,
реализацию TLS-сервера внутри ядра RicGate и многих других задач. Данная
библиотека широко распространена среди разработчиков использующих C++.
Для
автоматизации
сборки
проекта
RicGate
была
выбрана
кроссплатформенная система CMake. Это система позволяет ускорить и
автоматизировать процесс сборки проекта.
Разработка проекта велась в среде Clion 2017.2. Помимо удобства
разработки на языке C++ данная платформа имеют возможности по интеграции с
Boost и CMake.
4.2 Описание реализации
Реализация RicGate базировалась на диаграммах UML, описанных в 3
главе. Приложение А содержит в себе листинг кода разработанного сервиса.
Сервис разрабатывался так, чтобы ядро программы оставалось максимально
неизменным, а весь новый функционал был заключен внутри плагинов.
60
Ядро представляет собой класс управляющей маршрутизацией данных от
одних плагинов к другим и контролем подключений объектов.
Реализация базировалась на комбинации использования стандартных
библиотек с использование библиотек Boost.
Так для внутреннего логирования используется библиотека Booost.log. При
работе сервиса информация о его деятельности пишется в специальный файл,
который хранится в отдельной директории. При этом каждый день создается
новый файл. Это необходимо чтобы отлаживать работу плагинов и для
дополнительного мониторинга как самих действий объектов, так и управляющих
воздействий.
Формат
строки
логов:
[«Время»]
[«тип
сообщения»]
[«протокол»#«идентификатор объекта»]: «сообщение».
Для запуска сервиса необходимо написать конфигурационный файл,
содержащий в себе информацию о списке используемых плагинов и
дополнительные настройки к ним, например информацию о номера порта к
которому будут происходить подключения. Пример такого файла предоставлен в
приложении А. Список настроек может меняться индивидуально для каждого
плагина. Конфигурация выходных плагинов отличается тем, что в ней
содержится информация о тех входных плагинах, которые будут настроены на
выходной.
Сервис может работать как фоновый процесс и в оконном режиме, с
выводом критических сообщений из логирования.
4.3 Добавление плагинов
Входные и выходные плагины создаются на базе двух классов Inlet и Outlet
соответственно, данные названия соответствуют их назначению. Для создание
нового
плагина
достаточно
создать
новый
класс
унаследованный
от plugin_api::Inlet и вызвать макрос DECLARE_PLUGIN, который экспортирует
его.
61
Рассмотрим создание новых плагинов каждого из типов на основе
приложения А. Каждый родительский класс для плагинов наследуется от одного
и того же класса Interface. Для них в этом классе объявляются общие свойства
доступные из ядра RicGate:
– идентификатор;
– тип плагина (Inlet или Outlet);
– тип используемого протокола (TCP, UDP).
Так же объявляются общие методы:
– stop – остановка плагина;
– loadConfiguration – обработка конфигурационного файла.
4.3.1 Входной плагин
Для создания нового плагина необходимо наследоваться от класса Inlet.
Для
этого
класса
основными
являются
следующие
обязательные
при
наследовании методы:
–
startConnection – метод, который вызывается в плагине, после
установления соединения с объектом контроля;
–
processRead – обработчик данных, вызывается при получении данных
от объекта. В нем закладывается логика разбора и обработки данных от объектов
контроля;
–
processWrite – метод, который вызывается после отправки данных на
устройство;
–
processCommand – обработчик команды полученной от выходного
плагина (Outlet).
В этом классе так же объявлена ссылка на интерфейс для вызова методов
заданных в ядре – api_. Данный интерфейс определяется в ядре и позволяет
вызывать методы, которое отправляют данные, как на контролируемое
устройство, так и в выходной плагин, управляют подключением объекта.
62
Такая простая схема позволяет создавать новые плагины, начиная с
минимального шаблона, расширяя их по мере необходимости, не изменяя ядро
сервиса.
4.3.2 Выходной плагин
Принцип создания выходного плагина те же. Остается интерфейс api_
только с другими полями, свойства остаются те же, остается метод
loadConfiguration. Меняются только методы в соответствии с назначением
плагина:
–
sendDataOut – обработать отправить данные полученные от объекта;
–
sendStatus – отправить информацию о статусе объекта контроля;
–
sendRpc – отправить rpc-запрос (например, авторизация объекта).
По сути каждый выходной плагин – маленькая программа, слабо
зависимая от ядра RicGate. Выходной плагин инициализируется один раз и в
единственном экземпляре. После создания ядро никак не управляет выходными
плагинами, поэтому в них нужно предусмотреть механизм внутреннего контроля
своего состояния (например, контроль подключения к стороннему сервису и
механизм его восстановления в случае сбоев и т.п.).
4.3.3 Сборка сервиса
Как
упоминалось
ранее
для
сборки
проекта
используется
кроссплатформенная система CMake. При ее использовании сборка проекта
достаточно использовать команду make в директории с проектом, чтобы RicGate
и его плагины начали компилироваться. Для указаний правил сборки, проверки
необходимых библиотек и создания других правил пишется специальный файл с
названием CMakeLists.txt. Файл для сервиса и пример файла для плагина есть в
Приложении A.
63
Заключение
Результатом
выпускной
квалификационной
работы
является
разработанный расширяемый сетевой сервис межпротокольного взаимодействия
для высоконагруженных телематических систем с асинхронной сервисориентированной архитектурой. Была исследована область применении, как
телематических систем, так и разработанного сервиса. Были изучена предметная
область с целью определения требования к сервису и поиску его аналогов.
Выполнено его проектирование. Описана логика взаимодействия с сервисом.
Описан способ его расширения путем добавления новых плагинов.
Данный
взаимодействия
проект
без
позволяет
внедрения
расширять
в
основную
список
межпротокольного
программу,
поддерживает
возможность своего масштабирования. Асинхронная архитектура позволяет
работать
с
высоконагруженными
системами.
Реализация
сервис-
ориентированной архитектуры позволяет внедрять RicGate в любые системы
поддерживающие такую архитектуру.
Значит,
поставленная
цель
выпускной
достигнута, а поставленные задачи выполнены.
квалификационной
работы
64
Список использованной литературы
1
Комаров, В.В. Архитектура и стандартизация телематических и
интеллектуальных транспортных систем. Зарубежный опыт и отечественная
практика. / В.В. Комаров, С.А. Гараган. – М.: НТБ «Энергия», 2012. – 352с.
2
Нерсесов, Д. Телематика – новое слово в автомобильной электронике. /
Д. Нерсесов // Беспроводные технологии. – 2010. – №2. – С.40–46.
3
Ермак, С. Высоко чувство / С. Ермак // Эксперт Урал. – 2014. – №22. –
С.30–35.
4
Волкова, В.Н. Теория систем: Учеб. Пособие / В.Н. Волкова, А.А.
Денисов. – М.: Высш. шк., 2006. – 679с.
5
Антонов, А.В. Системный анализ. Учеб. для вузов / А.В. Антонов. – М.:
Высш. шк., 2004. – 454с.
6
Скалозуб, В.В. Прикладной системный анализ интеллектуальных систем
транспорта / В. В. Скалозуб, В. М. Ильман. – Д.: Изд–во Днепропетр. нац. ун–та
ж.–д. трансп. им. акад. В. Лазаряна, 2013. – 221с.
7
Липаев, В.В., Программная инженерия. Методологические основы. /
В.В. Липаев. – М.: ТЕИС, 2006. – 608с.
8
Комаров, В.В. Интеллектуальные задачи телематических транспортных
систем и интеллектуальная транспортная система / В.В. Комаров // T–Comm –
Телекоммуникации и Транспорт выпуск. – 2012. – №4. – С.34–38.
9
Лахтина, Н. Техническое обеспечение телематических систем: учеб.
пособие. Часть 3 / Н.Ю. Лахтина, К.Г. Манушакян. – М.: МАДИ, 2013. – 52с.
10 ГОСТ Р ИСО 14813–1–2011 Интеллектуальные транспортные системы.
Схема построения архитектуры интеллектуальных транспортных систем. Часть
1. Сервисные домены в области интеллектуальных транспортных систем,
сервисные группы и сервисы. – М.: Стандартинформ, 2011. – 25с.
65
11 National Intelligent Transportation System (ITS) Architecture. Executive
Summary. Research and Innovation Technology Administration (RITA). –
Washington D.C.: – US Department of Transportation., 2007. – 27c.
12 Конькова, Н. Инновационные интернет–технологии в деятельности
хозяйствующих субъектов: Интернет вещей: [Электронный ресурс]. – Режим
доступа:
http://docplayer.ru/43473766–Innovacionnye–internet–tehnologii–v–
deyatelnosti–hozyaystvuyushchih–subektov–internet–veshchey.html. – Дата доступа:
01.06.2017
13 Linda, M. Car–hacking: Remote Access and Other Security Issues:
[Электронный
ресурс].
–
Режим
доступа:
http://www.computerworld.com/article/2505402/security0/car–hacking––remote–
access–and–other–security–issues.html. – Дата доступа: 01.06.2017
14 Коптелов, А.
Сервис–ориентированная архитектура: от концепции к
применению. / А. Коптелов, В. Голубев // Byte Россия. – 2008. – №6. – С.25–28.
15 Хаммел, Р. Последовательная передача данных / Р. Хаммел. – М.: Мир,
1996. – с. 752
16 Torjo, J. Boost.Asio C++ Network Programming / J.Torjo. – Birmingham,
UK: Packt Publishing Ltd, 2013. – 359 c.
17 Маквитти, Л. Архитектура SOA как она есть / Л. Маквитти // Сети и
системы связи – 2006. – №2 – C.50–53.
18 Кнут, Д. Искусство программирования, том 1. Основные алгоритмы – 3–
е изд. / Д. Кнут. – М.: «Вильямс», 2006. – 720с.
19 Лахтина, Н. Техническое обеспечение телематических систем: учеб.
пособие. Часть 4 / Н.Ю. Лахтина, К.Г. Манушакян. – М.: МАДИ, 2013. – 52с.
20 Ayanoglu, Е. Mastering RabbitMQ. / E. Ayanoglu, Y. Aytaş, D. Nahum. –
Birmingham, UK: Packt Publishing Ltd, 2013. – 286c.
21 Грингард, С. Интернет вещей. Будущее уже здесь. / С. Грингард. – М.:
ООО «Альпина Паблишер», 2016. – 188с.
66
22 Rajive, J. The Industrial Internet of Things Volume G5: Connectivity
Framework: [Электронный ресурс] // Industrial internet consortium. – Режим
доступа: https://www.iiconsortium.org/ pdf/IIC_PUB_G5_V1.0_PB_20170228.pdf . –
Дата доступа: 01.06.2017
23 Сысоева, С. Новые системы телематики. Максимум безопасности,
надежности и беспроводных сервисных услуг / С. Сысоева // Компоненты и
технологии. – 2011. – №11. – C.29–38.
24 Приказ Минтранса РФ от 31 июля 2012 г. № 285 "Об утверждении
требований к средствам навигации, функционирующим с использованием
навигационных
сигналов
предназначенным
для
системы
ГЛОНАСС
обязательного
или
оснащения
ГЛОНАСС/GPS
транспортных
и
средств
категории М, используемых для коммерческих перевозок пассажиров, и
категории
N,
используемых
для
перевозки
опасных
грузов".
–
М.:
Стандартинформ, 2012. – 82c.
25 ISO/IEC 14882:2014 – Information technology – Programming languages –
C++. – Washington, DC: American National Standards Institute, 2014. – 809с.
26 Сик, Д. C++ Boost Graph Library / Д.Сик, Л.-К. Ли. – СПб.: Питер, 2006. –
304с.
27 Страуструп, Б. Язык программирования C++ – 3-е изд. / Б. Страуструп.
– М.: «Издательство БИНОМ», 1999г. – 991с.
28 Савичев, Д.В. Основные проблемы интернета вещей / Д.В. Савичев //
Естественнонаучные, инженерные и экономические исследования в технике,
промышленности, медицине и сельском хозяйстве: материалы I молодёжной
научно-практической конференции с международным участием; под общ. ред. С.
Н. Девицыной. – Белгород: ИД "Белгород", 2017. – С.252-255.
67
Приложение А
(обязательное)
Фрагменты кода
A.1 Main.cpp – точка входа RicGate
#include "config.hpp"
#include "Core.hpp"
#include <iostream>
#include <string>
#include <boost/application.hpp>
#include <boost/application/context.hpp>
#include <boost/application/aspects/process_id.hpp>
#include <boost/make_shared.hpp>
#include <boost/program_options.hpp>
using namespace std;
int main(int argc, char* argv[])
{
namespace po = boost::program_options;
namespace app = boost::application;
po::options_description desc("Options");
desc.add_options()
("help,h", "Print help message")
("foreground,F", "foreground mode");
po::variables_map vm;
try {
po::store(po::command_line_parser(argc, argv).options(desc).run(),vm);
/* – – help or – h */
68
if (vm.count("help"))
{
std::cout << desc << std::endl;
return EXIT_SUCCESS;
}
po::notify(vm);
}
catch (const std::exception& e) {
std::cerr << e.what() << std::endl;
std::cout << desc << std::endl;
std::cerr << boost::diagnostic_information(e) << std::endl;
return EXIT_FAILURE;
}
app::context appContext;
appContext.insert<app::path>(
std::make_shared<app::default_path>());
appContext.insert<app::args>(
std::make_shared<app::args>(argc, argv));
app::auto_handler<RicGate::Core> appCore(appContext);
int ret = 0;
boost::system::error_code ec;
if (vm.count("foreground")) {
ret = app::launch<app::common>(appCore, appContext, ec);
}
else {
ret = app::launch<app::server>(appCore, appContext, ec);
}
if (ec) {
std::cerr << "Error message: " << ec.message()
69
<< " (" << ec.value() << ")" << std::endl;
ret = EXIT_FAILURE;
}
return ret;
}
A.2 Core.hpp – ядро сервиса
#ifndef RIC_GATE_CORE_HPP
#define RIC_GATE_CORE_HPP
#include "log.hpp"
#include "PluginLoader.hpp"
#include "TcpServer.hpp"
#include "UdpServer.hpp"
#include "Timer.hpp"
#include <vector>
#include <boost/application/context.hpp>
#include <boost/application/aspects/process_id.hpp>
#include <boost/program_options/variables_map.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/log/utility/setup/settings.hpp>
namespace RicGate {
class TcpServer;
class PluginLoader;
class Core {
public:
/// Core constructor.
Core(boost::application::context &context);
/// Core destructor.
70
~Core();
/// Handles run signal.
int operator()();
/// Handles stop signal.
bool stop();
boost::application::process_id::native_pid_t getPid() const
{ return context_.find<boost::application::process_id>()– >pid(); }
/// io_service getter.
boost::shared_ptr<boost::asio::io_service> const& getIos() const
{ return ioService_; }
boost::shared_ptr<PluginLoader> const& getPluginLoader() const
{ return pluginLoader_; }
private:
/// Runner ric– gate deamon
void run();
/// load settinfs frim configuration file
void loadConfiguration();
/// logger attributes
logging::logger log_;
/// application context.
boost::application::context& context_;
/// variables map.
boost::program_options::variables_map& variablesMap;
/// Boost.Asio.
boost::shared_ptr<boost::asio::io_service> ioService_;
/// Plugin factory
boost::shared_ptr<PluginLoader> pluginLoader_;
/// Incoming TCP connections acceptors.
std::vector<boost::weak_ptr<TcpServer> > tcpListeners_;
71
/// Incoming UDP connections receiver;
std::vector<boost::weak_ptr<UdpServer> > udpReceivers_;
/// Configuration tree.
boost::property_tree::ptree configuration_;
/// Flag for second stopping core
bool isStopping;
};
}
#endif
A.3 Core.cpp – ядро сервиса
#include "Core.hpp"
#include <boost/asio/io_service.hpp>
#include <boost/asio/ip/host_name.hpp>
#include <boost/bind.hpp>
#include <boost/container/flat_set.hpp>
#include <boost/foreach.hpp>
#include <boost/filesystem/fstream.hpp>
#include <boost/log/attributes.hpp>
#include <boost/log/utility/setup/settings.hpp>
#include <boost/log/utility/setup/from_settings.hpp>
#include <boost/log/utility/setup/formatter_parser.hpp>
#include <boost/log/utility/setup/filter_parser.hpp>
#include <boost/make_shared.hpp>
#include <boost/property_tree/json_parser.hpp>
#include <boost/application/aspects/path.hpp>
#include <config.hpp>
#include <Dptree.hpp>
72
namespace RicGate
{
namespace app = boost::application;
void initLogger()
{
namespace attrs = boost::log::attributes;
BOOST_AUTO(log_core, boost::log::core::get());
log_core– >add_global_attribute("TimeStamp", attrs::local_clock());
log_core– >add_global_attribute("RecordID", attrs::counter<unsigned long>());
boost::log::register_simple_formatter_factory<logging::severity,
char>("Severity");
boost::log::register_simple_filter_factory<logging::severity, char>("Severity");
boost::log::settings log_settings;
//log_settings["Core.Filter"] = "%Severity% >= normal";
log_settings["Sinks.Console.Destination"] = "Console";
log_settings["Sinks.Console.Format"]
=
"[%TimeStamp%]
<%Severity%>\t[%Channel%] – %Message%";
log_settings["Sinks.Console.AutoFlush"] = true;
log_settings["Sinks.Console.Asynchronous"] = false;
boost::log::init_from_settings(log_settings);
}
Core::Core(app::context& context)
: log_(boost::log::keywords::channel = "core")
, context_(context)
, variablesMap(*context_.find<boost::program_options::variables_map>())
, isStopping(false)
{}
73
Core::~Core()
{
BOOST_LOG_SEV(log_, logging::info) << "Closing...";
BOOST_AUTO(logging_core, boost::log::core::get());
logging_core– >flush();
logging_core– >remove_all_sinks();
}
int Core::operator()()
{
int ret = EXIT_SUCCESS;
try {
initLogger();
loadConfiguration();
run();
}
catch (...) {
BOOST_LOG_SEV(log_, logging::warning)
<< boost::current_exception_diagnostic_information();
stop();
ret = EXIT_FAILURE;
}
boost::log::core::get()– >flush();
return ret;
}
bool Core::stop()
{
BOOST_LOG_SEV(log_, logging::warning)
<<
"Shutting
down
after
boost::current_exception_diagnostic_information();
a
catch"
<<
74
boost::log::core::get()– >flush();
if (isStopping){
BOOST_LOG_SEV(log_, logging::info) << "Second stopping core";
if (ioService_) {
ioService_– >stop();
}
boost::log::core::get()– >flush();
return true;
}
isStopping = true;
if (!ioService_) return true;
boost::asio::io_service::work work(*ioService_);
BOOST_LOG_SEV(log_, logging::flood) << "Freeing listeners";
BOOST_FOREACH(const
boost::weak_ptr<TcpServer>&
listener,
tcpListeners_) {
BOOST_AUTO(p, listener.lock());
if (p) {
p– >closeAccept();
}
}
BOOST_FOREACH(const
boost::weak_ptr<UdpServer>&
udpReceivers_) {
BOOST_AUTO(p, receiver.lock());
if (p) {
p– >closeReceive();
}
}
pluginLoader_– >unload();
BOOST_LOG_SEV(log_, logging::flood)
receiver,
75
<< "Cleanup is done";
boost::log::core::get()– >flush();
return true;
}
void readConfig(logging::logger& log_, boost::log::settings& logConfig, const
boost::property_tree::ptree& config, std::string path="")
{
BOOST_FOREACH(const boost::property_tree::ptree::value_type& v, config) {
if (v.second.empty()) {
logConfig[path + v.first] = v.second.data();
}
else {
readConfig(log_, logConfig, v.second, path + v.first + ".");
}
}
}
void Core::loadConfiguration()
{
namespace json_parser = boost::property_tree::json_parser;
const std::string configFile = DEFAULT_CONFIG_PATH;
BOOST_LOG_SEV(log_, logging::notify)
<< "load configuration from " << configFile;
try {
json_parser::read_json(configFile, configuration_);
}
catch (const json_parser::json_parser_error& e) {
BOOST_LOG_SEV(log_, logging::critical)
<< "Configuration file parsing failed: " << e.what();
throw;
76
}
typedef boost::optional<boost::property_tree::ptree&> log_tree_t;
if (const log_tree_t log_tree = configuration_.get_child_optional("log")) {
boost::log::settings logSettings;
readConfig(log_, logSettings, *log_tree);
BOOST_AUTO(logCore, boost::log::core::get());
logCore– >flush();
logCore– >remove_all_sinks();
boost::log::init_from_settings(logSettings);
}
}
void Core::run()
{
BOOST_LOG_SEV(log_, logging::info)
<< "Working directory: " << boost::filesystem::current_path();
const std::string pidname = DEFAULT_PID_PATH;
boost::filesystem::path pidfile(pidname);
if (pidfile.is_relative()) {
pidfile = boost::filesystem::current_path() / pidfile;
}
if (boost::filesystem::exists(pidfile)){
boost::filesystem::ifstream is(pidfile);
boost::application::process_id::native_pid_t lastPid;
is >> lastPid;
}
boost::filesystem::ofstream os(pidfile, std::ios_base::trunc);
os << getPid();
std::size_t numThreads = std::max(1u, boost::thread::hardware_concurrency());
BOOST_LOG_SEV(log_, logging::info)
77
<< "Asio thread pool size: " << numThreads ;
ioService_ = boost::make_shared<boost::asio::io_service>(numThreads);
boost::asio::io_service::strand strand_(*ioService_);
boost::asio::ip::tcp::resolver resolver_(*ioService_);
boost::asio::ip::tcp::socket socket_(*ioService_);
RelativeTimer timer_(*ioService_);
pluginLoader_ = boost::make_shared<PluginLoader>(boost::ref(*this));
pluginLoader_–
>loadConfiguration(configuration_.get_child("plugin",emptyPtree<boost::property_tre
e::ptree>()));
InletControl& inletControl = pluginLoader_– >getInletControl();
InletMap inletMap = inletControl.getInletMap();
if (inletMap.size()==0) {
BOOST_LOG_SEV(log_, logging::info)
<< "loaded plugins is 0";
return;
}
SessionList_ session;
std::string address = "0.0.0.0";
try {
InletMap::const_iterator iter = inletMap.begin();
while (iter != inletMap.end()) {
session = *iter;
unsigned short port = session.port_;
BOOST_LOG_SEV(log_, logging::info)
<< session.protocol_ << " for " << session.puid_ << " – " << address <<
":" << port;
if (session.protocol_ == PluginApi::ProtocolType::TCP) {
78
boost::shared_ptr<TcpServer> tcpListener =
boost::make_shared<TcpServer>(boost::ref(*this), address, port);
tcpListeners_.push_back(tcpListener);
tcpListener– >startAccept();
}
else {
boost::shared_ptr<UdpServer> udpReceiver =
boost::make_shared<UdpServer>(boost::ref(*this), address, port);
udpReceivers_.push_back(udpReceiver);
udpReceiver– >startReceive();
}
iter++;
}
}
catch (const boost::system::system_error& e){
BOOST_LOG_SEV(log_, logging::critical)
<< e.what() << " (" << e.code().value() << ")";
throw;
}
if (numThreads > 1) {
boost::thread_group threads;
for (std::size_t i = 0; i < numThreads; ++i)
{
threads.create_thread(
boost::bind(&boost::asio::io_service::run, ioService_));
}
threads.join_all();
} else {
ioService_– >run();
79
}
isStopping = true;
pluginLoader_.reset();
boost::log::core::get()– >flush();
}
}
A.4 Connection.hpp – контроль соединений с объектами
#ifndef RIC_GATE_CONNECTION_HPP
#define RIC_GATE_CONNECTION_HPP
#include "log.hpp"
#include "Core.hpp"
#include "api/PluginInlet.hpp"
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/strand.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/weak_ptr.hpp>
#include <boost/make_shared.hpp>
namespace RicGate{
class Core;
class Connection
: public boost::enable_shared_from_this<Connection>
, private boost::noncopyable
{
80
public:
explicit Connection(Core const& core);
explicit Connection(Core const& core, boost::asio::ip::udp::endpoint endpoint);
public:
~Connection();
static boost::shared_ptr<Connection> create(Core const& core)
{
boost::shared_ptr<Connection> p = boost::make_shared<Connection>(core);
p– >weak_this_ = p;
return p;
}
static
boost::shared_ptr<Connection>
create(Core
const&
core,
boost::asio::ip::udp::endpoint endpoint)
{
boost::shared_ptr<Connection> p = boost::make_shared<Connection>(core,
endpoint);
p– >weak_this_ = p;
return p;
}
boost::asio::ip::tcp::endpoint remoteEndpoint() const
{ return
remoteEndpoint_; }
boost::asio::ip::tcp::socket& socket() { return socket_; }
public: boost::asio::ip::tcp::socket const& socket() const { return socket_; }
boost::asio::ip::udp::socket& socketUdp() { return socketUdp_; }
public: boost::asio::ip::udp::socket const& socketUdp() const { return socketUdp_;
}
bool authId(const std::string& id);
/*TCP*/
void onConnection();
81
void tryStrand(boost::function<void()> f);
void postStrand(boost::function<void()> f);
void dispatchStrand(boost::function<void()> f);
void close();
/* UDP*/
void onUdpConnection();
RelativeTimerWrapper createRelativeTimer();
void closeUdp();
void fillUdpBuffer(std::vector<unsigned char> buf) {
udpBuf.clear();
udpBuf = buf;}
boost::weak_ptr<PluginApi::Inlet> const& apiInlet() const { return apiInlet_; }
bool getReestablished(){
return reestablished_;
}
void toggleReestablished(){
BOOST_LOG_SEV(log_, logging::debug)
<< "Toggle";
reestablished_ = !reestablished_;
}
void setTimer(std::size_t t)
{
BOOST_LOG_SEV(log_, logging::debug)
<< "Set timer";
if (!timer_) {
timer_ = createRelativeTimer();
}
timer_– >expires_from_now(boost::chrono::seconds(t));
82
timer_–
>async_wait(boost::bind(&Connection::toggleReestablished,
shared_from_this()));
}
private:
boost::optional<RelativeTimerWrapper> timer_;
void
processRead(boost::shared_ptr<PluginApi::Inlet>
const&
processInlet,
const boost::system::error_code& errorCode, std::size_t readSize);
void processUdpRead(boost::shared_ptr<PluginApi::Inlet> const& processInlet,
const boost::system::error_code& errorCode, std::size_t readSize);
void processWrite(boost::shared_ptr<PluginApi::Inlet> const& processInlet,
const boost::system::error_code& errorCode, bool isCommand);
logging::logger log_;
Core const& core_;
boost::shared_ptr<PluginLoader> pluginLoader;
boost::shared_ptr<boost::asio::io_service> ioService;
boost::asio::io_service::strand strand_;
/// Socket for the connection.
boost::weak_ptr<Connection> weak_this_;
boost::asio::ip::tcp::socket socket_;
boost::asio::ip::udp::socket socketUdp_;
boost::weak_ptr<PluginApi::Inlet> apiInlet_;
boost::asio::ip::tcp::endpoint remoteEndpoint_;
boost::asio::ip::udp::endpoint remoteEndpointUdp_;
boost::shared_ptr<PluginLoader> pluginLoader_;
bool alive_;
bool reestablished_;
boost::optional<std::string> id_;
std::vector<unsigned char> udpBuf;
83
/// Start the first asynchronous operation for the connection.
friend class tcpServer;
friend class udpServer;
friend class OutletControl;
};
typedef boost::shared_ptr<Connection> connectionPtr;
}
#endif
A.5 Connection.cpp – контроль соединений с объектами
#include "Connection.hpp"
#include <boost/foreach.hpp>
#include <boost/log/attributes.hpp>
#include <boost/log/utility/manipulators/dump.hpp>
namespace RicGate
{
Connection::Connection(Core const& core)
: log_(boost::log::keywords::channel = "connection")
, core_(core)
, ioService(core_.getIos())
, strand_(*ioService)
, socket_(*ioService)
, socketUdp_(*ioService)
, pluginLoader_(core_.getPluginLoader())
, alive_(true)
, reestablished_(false)
{
/* TCP connection */
84
}
Connection::Connection(Core const& core, boost::asio::ip::udp::endpoint endpoint)
: log_(boost::log::keywords::channel = "connection")
, core_(core)
, ioService(core_.getIos())
, strand_(*ioService)
, socket_(*ioService)
, socketUdp_(*ioService, endpoint)
, alive_(true)
, reestablished_(false)
{
/* For UDP connection*/
}
Connection::~Connection()
{
BOOST_LOG_SEV(log_, logging::info) << "Session destroyed";
boost::log::core::get()– >flush();
if (id_) {
pluginLoader_– >getOutletControl().deleteSession(*id_, weak_this_);
}
}
void Connection::onConnection()
{
boost::system::error_code ignored;
boost::asio::ip::tcp::endpoint localEndpoint = socket_.local_endpoint(ignored);
remoteEndpoint_ = socket_.remote_endpoint(ignored);
BOOST_AUTO(pluginApi, boost::make_shared<PluginApi::ApiInlet>());
auto self = shared_from_this();
85
pluginApi– >readAtLeast = [&](boost::asio::streambuf& buf, std::size_t minByte)
{
BOOST_LOG_SEV(log_, logging::debug) << "readAtLeast(): " << minByte;
if (!alive_) {
return;
}
boost::asio::async_read(socket_, buf, boost::asio::transfer_at_least(minByte),
strand_.wrap(
boost::bind(&Connection::processRead,
shared_from_this(),
apiInlet_.lock(), _1, _2)));
};
pluginApi– >readUntil = [&](boost::asio::streambuf& buf, const std::string&
untilBytes) {
BOOST_LOG_SEV(log_, logging::debug) << "readUntil()";
if (!alive_) {
return;
}
boost::asio::async_read_until(socket_, buf, untilBytes, strand_.wrap(
boost::bind(&Connection::processRead,
shared_from_this(),
apiInlet_.lock(), _1, _2)));
};
pluginApi– >readSome = [&](const boost::asio::mutable_buffer& buf) {
BOOST_LOG_SEV(log_, logging::debug) << "readSome()";
if (!alive_) {
return;
}
socket_.async_read_some(boost::asio::mutable_buffers_1(buf),
strand_.wrap
(boost::bind(&Connection::processRead, shared_from_this(), apiInlet_.lock(), _1,
_2)));
86
};
pluginApi– >writeData = [&](const boost::asio::const_buffer& buf, bool
isCommand) {
BOOST_LOG_SEV(log_,
logging::debug)
<<
"writeData():
"
<<
boost::asio::buffer_size(buf) << " bytes";
if (!alive_) {
return;
}
boost::asio::async_write(socket_,
boost::asio::const_buffers_1(buf),
strand_.wrap(
boost::bind(&Connection::processWrite,
shared_from_this(),
apiInlet_.lock(), _1, isCommand)));
};
pluginApi– >dispatchStrand = boost::bind(&Connection::dispatchStrand, this,
_1);
pluginApi–
>createRelativeTimer
=
boost::bind(&Connection::createRelativeTimer, this);
pluginApi–
>closeConnection
=
boost::bind(&Connection::close,
shared_from_this());
InletControl& inletControl = pluginLoader_– >getInletControl();
SessionList_ sessionInfo = inletControl.getInlet(localEndpoint.port());
pluginLoader_–
getOutletControl().bindConnectionRoute(sessionInfo.puid_,*pluginApi,
sessionInfo.port_, weak_this_);
boost::shared_ptr<PluginApi::Inlet>processInlet=InletControl.create(
sessionInfo.puid_);
apiInlet_ = processInlet;
processInlet– >setupApi(pluginApi);
try {
>
87
processInlet– >startConnection(sessionInfo.mode_);
} catch (...) {
BOOST_LOG_SEV(log_,
logging::critical)
<<
"Exception
in
inlet
startConnection() "<< boost::current_exception_diagnostic_information();
close();
}
}
void Connection::onUdpConnection()
{
boost::system::error_code ignored;
boost::asio::ip::udp::endpoint
localEndpoint
=
socketUdp_.local_endpoint(ignored);
remoteEndpointUdp_ = socketUdp_.remote_endpoint(ignored);
BOOST_LOG_SEV(log_, logging::info) << "Udp connection accepted";
BOOST_AUTO(pluginApi, boost::make_shared<PluginApi::ApiInlet>());
auto self = shared_from_this();
pluginApi– >udpRead = [&](const boost::asio::mutable_buffer& buf) {
if (!alive_)
return;
socketUdp_.async_receive_from(boost::asio::mutable_buffers_1(buf),
remoteEndpointUdp_,
(boost::bind
(&Connection::processUdpRead,
shared_from_this(), apiInlet_.lock(), _1, _2)));
};
pluginApi– >writeUdpData = [this, self](const boost::asio::const_buffer& buf,
bool isCommand) {
if (!alive_)
return;
88
socketUdp_.async_send_to(boost::asio::buffer(buf),
remoteEndpointUdp_,boost::bind(&Connection::processWrite,
shared_from_this(),
apiInlet_.lock(), _1, isCommand));
};
pluginApi–
>closeConnection
=
boost::bind(&Connection::closeUdp,
shared_from_this());
//TODO route need
InletControl& inletControl = pluginLoader_– >getInletControl();
SessionList_ sessionInfo = inletControl.getInlet(localEndpoint.port());
boost::shared_ptr<PluginApi::Inlet>
processInlet
=
inletControl.create(sessionInfo.puid_);
apiInlet_ = processInlet;
processInlet– >setupApi(pluginApi);
try {
processInlet– >startConnection(sessionInfo.mode_);
} catch (...) {
BOOST_LOG_SEV(log_, logging::critical) << "Exception in udpInlet"
<< boost::current_exception_diagnostic_information();
close();
}
}
void
Connection::processRead(boost::shared_ptr<PluginApi::Inlet>
const&
processInlet, const boost::system::error_code& errorCode, std::size_t readSize)
{
BOOST_LOG_SEV(log_, logging::debug) << "processRead()" << readSize;
if (errorCode) {
if (errorCode == boost::asio::error::eof) {
BOOST_LOG_SEV(log_, logging::info) << "ERROR: Connection has been
closed by inlet";
89
if (timer_) {
timer_– >cancel();
}
return;
}
if (errorCode == boost::asio::error::operation_aborted) {
BOOST_LOG_SEV(log_, logging::info) << "ERROR: Connection has been
closed, operation aborted";
if (timer_) {
timer_– >cancel();
}
// close();
return;
}
BOOST_LOG_SEV(log_,
logging::error)
<<
"Reading
failed:
"
<<
errorCode.message() << " (" << errorCode.value() << ")";
// close();
if (timer_) {
timer_– >cancel();
}
return;
}
BOOST_LOG_SEV(log_, logging::debug) << "readData (" << readSize << ")";
try {
processInlet– >processRead(readSize);
} catch (...) {
BOOST_LOG_SEV(log_, logging::critical) << "Exception in processRead():
"<< boost::current_exception_diagnostic_information();
close();
90
}
}
void
Connection::processUdpRead(boost::shared_ptr<PluginApi::Inlet>
const&
processInlet, const boost::system::error_code& errorCode, std::size_t readSize)
{
BOOST_LOG_SEV(log_, logging::debug) << "processRead()" << readSize;
if (errorCode) {
if (errorCode == boost::asio::error::eof) {
BOOST_LOG_SEV(log_, logging::info) << "ERROR: Connection has been
closed by inlet";
return;
}
if (errorCode == boost::asio::error::operation_aborted) {
BOOST_LOG_SEV(log_, logging::info) << "ERROR: Connection has been
closed, operation aborted";
return;
}
BOOST_LOG_SEV(log_,
logging::error)
<<
"Reading
failed:
"
<<
errorCode.message() << " ("<< errorCode.value() << ")";
return;
}
BOOST_LOG_SEV(log_, logging::debug) << "readData (" << readSize << ")";
try {
processInlet– >processRead(readSize);
} catch (...) {
BOOST_LOG_SEV(log_, logging::critical) << "Exception in processRead():
"<< boost::current_exception_diagnostic_information();
closeUdp();
}
91
}
void
Connection::processWrite(boost::shared_ptr<PluginApi::Inlet>
const&
processInlet, const boost::system::error_code& errorCode, bool isCommand)
{
BOOST_LOG_SEV(log_, logging::debug) << "processWrite()";
if (errorCode) {
BOOST_LOG_SEV(log_, logging::error) << "processWrite() failed: " <<
errorCode.message() << " ("<< errorCode.value() << ")";
return;
}
try {
processInlet– >processWrite(isCommand);
} catch (...) {
BOOST_LOG_SEV(log_, logging::critical) << "Exception in processWrite(): "
<< boost::current_exception_diagnostic_information();
close();
}
}
void Connection::tryStrand(boost::function<void()> f)
{
try {
f();
} catch (...) {
BOOST_LOG_SEV(log_, logging::critical) << "Exception in handler executed
in strand: "
<< boost::current_exception_diagnostic_information();
close();
}
}
92
void Connection::postStrand(boost::function<void()> f)
{
strand_.post(boost::bind(&Connection::tryStrand,
shared_from_this(),
boost::move(f)));
}
void Connection::dispatchStrand(boost::function<void()> f)
{
strand_.dispatch(boost::bind(&Connection::tryStrand,
shared_from_this(),
boost::move(f)));
}
bool Connection::authId(const std::string& id)
{
BOOST_LOG_SEV(log_, logging::debug) << "authId()";
OutletControl& outlet = pluginLoader_– >getOutletControl();
if (id_) {
BOOST_LOG_SEV(log_, logging::info) << "New id " << id << "; id was " <<
id_;
outlet.changeSession(id);
return true;
}
if (outlet.addSession(id, shared_from_this())) {
id_ = id;
return true;
}
close();
return false;
}
void Connection::close()
93
{
alive_ = false;
if (timer_) {
timer_– >cancel();
}
BOOST_LOG_SEV(log_, logging::info) << "Closing connection";
boost::system::error_code ignored_ec;
socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignored_ec);
socket_.close();
}
void Connection::closeUdp()
{
alive_ = false;
if (timer_) {
timer_– >cancel();
}
BOOST_LOG_SEV(log_, logging::flood) << "Closing connection";
boost::system::error_code ignored_ec;
socketUdp_.shutdown(boost::asio::ip::udp::socket::shutdown_both, ignored_ec);
socketUdp_.close();
}
RelativeTimerWrapper Connection::createRelativeTimer()
{
BOOST_LOG_SEV(log_, logging::debug) << "createRelativeTimer()";
RelativeTimerWrapper rtw;
BOOST_AUTO(timer,
boost::make_shared<RelativeTimer>(boost::ref(*ioService)));
94
/// TODO: store shared_ptr<io_service_> in timer wrapper
rtw.async_wait
=
boost::bind(&RelativeTimer::async_wait<const
RelativeTimerWrapper::wait_handler_type&>, timer, _1);
rtw.cancel = boost::bind(&RelativeTimer::cancel, timer);
rtw.expires_from_now = boost::bind(&RelativeTimer::expires_from_now, timer,
_1);
return rtw;
}
}
A.6 PlaginLoader.hpp – загрузчик плагинов
#ifndef RIC_GATE_PLUGIN_FACTORY_HPP
#define RIC_GATE_PLUGIN_FACTORY_HPP
#include "api/PluginInfo.hpp"
#include "log.hpp"
#include "InletControl.hpp"
#include "OutletControl.hpp"
#include <boost/application.hpp>
#include <boost/function.hpp>
#include <boost/container/flat_map.hpp>
namespace RicGate {
class Core;
class PluginLoader
:private boost::noncopyable
{
public:
//Constructor
PluginLoader(Core& core);
95
//Destructor
~PluginLoader();
// Load plugin from pathDir
Puid load(const boost::filesystem::path& pathDir);
// Search dir for plugin and load() them
void loadFromDir(const boost::filesystem::path& pathDir);
// Create interface for a plugin (as Plugin Api)
PluginInterfacePtr create(Puid puid);
void loadConfiguration(const boost::property_tree::ptree& configuration);
void unload();
InletControl& getInletControl() { return inletControl_; }
OutletControl& getOutletControl() { return outletControl_; }
private:
// Logger file
logging::logger log_;
// TODO: Replace with dependency injector.
Core & core_;
// InletControl module
InletControl inletControl_;
// OutletControl module
OutletControl outletControl_;
// Outlets puid and ptr on Info of plugin
boost::container::flat_map<Puid, PluginInfoPtr> loadedPlugins_;
};
}
#endif
A.7 PlaginLoader.cpp – загрузчик плагинов
96
#include "PluginLoader.hpp"
#include "Core.hpp"
#include "Dptree.hpp"
#include <boost/foreach.hpp>
#include <boost/log/utility/manipulators/add_value.hpp>
#include <boost/throw_exception.hpp>
namespace RicGate {
PluginLoader::PluginLoader(Core& core)
: log_(boost::log::keywords::channel = "PluginLoader")
, core_(core)
, outletControl_(core_)
{
}
PluginLoader::~PluginLoader()
{
BOOST_LOG_SEV(log_, logging::info) << "Destroying";
boost::log::core::get()– >flush();
}
void
PluginLoader::loadConfiguration(const
boost::property_tree::ptree&
configuration)
{
using boost::property_tree::ptree;
BOOST_AUTO(pluginsPath, configuration.get_child_optional("path"));
if (pluginsPath) {
BOOST_FOREACH(const ptree::value_type& path, *pluginsPath) {
boost::filesystem::path path_(path.second.get<std::string>(""));
if (path_.is_absolute())
{
97
loadFromDir(path_);
}
else {
loadFromDir(boost::filesystem::current_path() / path_);
}
}
}
else {
loadFromDir(boost::filesystem::current_path());
}
inletControl_.loadConfiguration(configuration.get_child("inlet"));
outletControl_.loadConfiguration(configuration.get_child("outlet"));
}
Puid PluginLoader::load(const boost::filesystem::path &pathDir)
{
BOOST_AUTO(pinfo, boost::make_shared<PluginInfo>(pathDir));
const Puid puid = pinfo– >puid();
const PluginApi::ProtocolType protocolType = pinfo– >protocolType();
const
std::string
protocolTypeString
=
protocolType
PluginApi::ProtocolType::TCP ? "TCP" : "UDP";
BOOST_LOG_SEV(log_, logging::debug)
<< "Plugin loaded uid: " << puid
<< " name: " << pinfo– >name()
<< " protocol: " << protocolTypeString;
BOOST_AUTO(it, loadedPlugins_.find(puid));
if (it != loadedPlugins_.end()) {
BOOST_LOG_SEV(log_, logging::error)
<< "Plugin with same uid='" << puid << "' is already loaded"
" from " << it– >second– >library().location();
==
98
return "";
}
switch (pinfo– >type()) {
case PluginApi::INLET_:
inletControl_.add(pinfo);
break;
case PluginApi::OUTLET_:
outletControl_.add(pinfo);
break;
default:
BOOST_LOG_SEV(log_, logging::warning)
<< "Unknown plugin type: " << puid << ". Unloading it.";
return "";
}
loadedPlugins_.emplace(puid, pinfo);
return puid;
}
void PluginLoader::loadFromDir(const boost::filesystem::path& pathDir)
{ //TODO mb need some refactor
BOOST_LOG_SEV(log_, logging::notify)
<< "Loading plugins from directory: " << pathDir;
boost::filesystem::path scan_dir(pathDir);
std::vector<boost::filesystem::path> files;
std::string plugins_;
std::copy(boost::filesystem::directory_iterator(scan_dir),
boost::filesystem::directory_iterator(), back_inserter(files));
BOOST_FOREACH(const boost::filesystem::path& path_name, files) {
99
const std::string extension
=
boost::algorithm::to_lower_copy(path_name.extension().string());
if (boost::dll::shared_library::suffix() == extension) {
BOOST_LOG_SEV(log_, logging::debug)
<< "Found plugin: " << path_name.string();
Puid pluginName = load(path_name);
plugins_.append(pluginName + "; ");
}
}
BOOST_LOG_SEV(log_, logging::info)
<< "Found plugins: " << plugins_;
}
PluginInterfacePtr PluginLoader::create(Puid puid)
{
BOOST_AUTO(it, loadedPlugins_.find(puid));
if (it != loadedPlugins_.end()) {
return (*it– >second)();
}
BOOST_LOG_SEV(log_, logging::error)
<< "Tried to instanciate not loaded plugin (uid: " << puid << ")";
std::out_of_range e("threre is no loaded plugin with such uid");
boost::throw_exception(e);
}
void PluginLoader::unload()
{
BOOST_LOG_SEV(log_, logging::info)
<< "Unload OutletControl";
100
outletControl_.remove();
BOOST_LOG_SEV(log_, logging::info)
<< "Unload InletControl";
inletControl_.remove();
/// TODO: Implement plugin_info holding by plugin instance
loadedPlugins_.clear();
}}
A.8 PluginApi.hpp – интерфейс доступа для плагинов
#ifndef PLUGIN_API_HPP
#define PLUGIN_API_HPP
#ifndef BOOST_ALL_DYN_LINK
# error "BOOST_ALL_DYN_LINK must be defined to work properly!"
#endif
#define MANGLING_MODE extern "C"
#define LIBRARY_API MANGLING_MODE BOOST_SYMBOL_EXPORT
#include <boost/function.hpp>
#include <boost/make_shared.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/shared_ptr.hpp>
#include "DptreeFwd.hpp"
namespace RicGate {
namespace PluginApi
{
// Callbacks
typedef boost::function<void(bool)> DataCallback;
typedef boost::function<void(boost::shared_ptr<dptree>)> CommandCallback;
101
// Plugin type enum.
enum PluginType {
INLET_,
OUTLET_
};
enum ProtocolType{
TCP,
UDP
};
// Plugin interface.
class Interface
{
public:
virtual ~Interface() {}
// Plugin type
virtual PluginType type() const = 0;
// Plugin uid
virtual const char* uid() const = 0;
// Plugin name
virtual const char* name() const = 0;
// Protocol type (Defenition need for connecting and interaction with Inlet)
virtual ProtocolType protocolType() const = 0;
// Plugin stop
virtual void stop() {}
// Plugin settings loader handler
virtual void loadConfiguration(const boost::property_tree::ptree&) {}
};
102
// Plugin factory function.
typedef boost::shared_ptr<Interface> create_shared();
} // PluginApi
typedef boost::shared_ptr<PluginApi::Interface> PluginInterfacePtr;
} // namespace RicGate
// Export plugin interface.
#define DECLARE_PLUGIN(T)
\
LIBRARY_API RicGate::PluginInterfacePtr
create_shared(void)
\
\
{ return boost::make_shared<T>(); }
\
// DECLARE_PLUGIN
#endif // PLUGIN_API_HPP
A.9 PluginInfo.hpp – базовый класс для плагинов
#ifndef PLUGIN_INFO_HPP
#define PLUGIN_INFO_HPP
#include "api/PluginApi.hpp"
#include <boost/dll.hpp>
#include <boost/lambda/bind.hpp>
#include <boost/lambda/construct.hpp>
#include <boost/typeof/typeof.hpp>
namespace RicGate {
typedef std::string Puid;
103
class PluginInfo
:private boost::noncopyable
{
public:
// Plugin info constructor.
explicit PluginInfo(const boost::filesystem::path& path_name)
: library_(path_name)
, creator_(library_.get<PluginApi::create_shared>("create_shared"))
{
BOOST_AUTO(instance, creator_());
type_ = instance– >type();
puid_ = instance– >uid();
name_ = instance– >name();
protocolType_ = instance– >protocolType();
}
// Overload call– operator with ability to construct plugin
PluginInterfacePtr operator()() const { return creator_(); }
// Plugin library
const boost::dll::shared_library& library() const { return library_; }
// Plugin type
PluginApi::PluginType type() const { return type_; }
// Plugin uid
Puid puid() const { return puid_; }
// Plugin name
std::string name() const { return name_; }
// Protocol type
PluginApi::ProtocolType protocolType() const { return protocolType_; }
104
private:
// Holds pointer to library which contains plugin
boost::dll::shared_library library_;
// Stores plugin constructor.
boost::function<PluginInterfacePtr()> creator_;
// Plugin type
PluginApi::PluginType type_;
// Plugin uid
Puid puid_;
// Plugin name
std::string name_;
// Plugin protocol
PluginApi::ProtocolType protocolType_;
};
typedef boost::shared_ptr<PluginInfo> PluginInfoPtr;
}
#endif
A.10 PluginInlet.hpp – базовый класс для входных плагинов
#ifndef PLUGIN_INLET_HPP
#define PLUGIN_INLET_HPP
#include "api/PluginApi.hpp"
#include "DptreeFwd.hpp"
#include "Timer.hpp"
#include <boost/asio/buffer.hpp>
#include <boost/asio/streambuf.hpp>
#include <boost/function.hpp>
namespace RicGate {
105
namespace PluginApi {
// Inlet api type.
typedef boost::function<void(bool)> CallbackAuthenticate;
struct ApiInlet {
// Read minimum or more bytes
boost::function<void(boost::asio::streambuf
&,
std::size_t
minByte)>
readAtLeast;
// While not untilBytes read data to the buffer
boost::function<void(boost::asio::streambuf &, const std::string &untilBytes)>
readUntil;
// Read data before end of flow
boost::function<void(const boost::asio::mutable_buffer &)> readSome;
// UDP read
boost::function<void(const boost::asio::mutable_buffer &)> udpRead;
// Data write
boost::function<void(const boost::asio::const_buffer &, bool isCommand)>
writeData;
// UDP send data
boost::function<void(const boost::asio::const_buffer &, bool isCommand)>
writeUdpData;
// Post callback in network strand.
boost::function<void(boost::function<void()>)> postStrand;
// Dispatch callback in network strand.
boost::function<void(boost::function<void()>)> dispatchStrand;
// Connection closet.
boost::function<void()> closeConnection;
// Authentication.
boost::function<void(std::string, CallbackAuthenticate)> authenticate;
// Send data to Outlet
106
boost::function<void(boost::shared_ptr<dptree>, DataCallback)> sendData;
// Send status to Outlet
boost::function<void(boost::shared_ptr<dptree>/*, DataCallbaсk */)> sendEvent;
// Send Rpc to web
boost::function<void(boost::shared_ptr<dptree>,
std::string
const&
name)>
sendRpc;
// create relative timer with wrapper
boost::function<RelativeTimerWrapper()> createRelativeTimer;
};
// Inlet plugin interface.
class Inlet
: public Interface {
public:
// Inlet plugin destructor.
virtual ~Inlet() { api_.reset(); }
// Inlet plugin type getter.
virtual PluginType type() const { return INLET_; }
// Inlet start handler.
virtual void startConnection(bool sess) = 0;
// Inlet read handler.
virtual void processRead(std::size_t bytes_transferred) = 0;
// Inlet write handler.
virtual void processWrite(bool isCommand) = 0;
// Inlet custom event .
virtual
bool
processCommand(boost::shared_ptr<dptree>
CommandCallback /*callback*/) = 0;
/*data*/,
107
// Inlet api setup handler.
void setupApi(boost::shared_ptr<ApiInlet> p) { api_ = p; }
protected:
// Inlet api.
boost::shared_ptr<ApiInlet> api_;
};
}//PluginApi
// Ptr to PluginApi::Inlet
typedef boost::shared_ptr<PluginApi::Inlet> PluginInletPtr;
} // namespace RicGate
#endif // PLUGIN_INLET_HPP
A.11 PluginOutlet.hpp – базовый класс для выходных плагинов
#ifndef PLUGIN_OUTLET_HPP
#define PLUGIN_OUTLET_HPP
#include "api/PluginApi.hpp"
#include "DptreeFwd.hpp"
namespace boost { namespace asio { class io_service; } }
namespace RicGate {
namespace PluginApi
{
struct ApiOutlet
{
boost::shared_ptr<boost::asio::io_service> ioService;
// send command
boost::function<bool(boost::shared_ptr<dptree>
callback)> sendCommand;
//
// send webReplies
data,
CommandCallback
108
//
boost::function<bool(boost::shared_ptr<dptree> data)> sendWebReply;
};
class Outlet
: public Interface
{
public:
virtual ~Outlet(){api_.reset();}
// Pluign type
virtual PluginType type() const { return OUTLET_;}
// Send data to Outlets
virtual void sendDataOut(boost::shared_ptr<dptree const> pt, DataCallback
callback) = 0;
// Update session status
virtual void sendStatus(boost::shared_ptr<dptree const> pt) = 0;
// Send web rpc
virtual void sendRpc(boost::shared_ptr<dptree const> pt, std::string const&
name) = 0;
// api for Outlet
void setupApi(boost::shared_ptr<ApiOutlet> p)
{
api_ = p;
}
// virtual void outletStop() = 0;
protected:
boost::shared_ptr<ApiOutlet> api_;
};
}//PluginOutlet
// Ptr to Outlet
typedef boost::shared_ptr<PluginApi::Outlet> OutletPtr;
109
}//RicGate
#endif
A.12 CMakeLists.txt – сборочный файл cmake
project(RicGate)
cmake_minimum_required(VERSION 2.8)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} – std=c++14")
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} – fPIC")
string(TOUPPER ${PROJECT_NAME} CONFIG_VAR_PREFIX)
set(CONFIG_VAR_PREFIX "${CONFIG_VAR_PREFIX}" CACHE INTERNAL
"Config variables prefix")
# Include required CMake modules
set(CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/libs/cmake)
include(utils)
include_directories(include)
#include_directories(.)
include_directories("${PROJECT_BINARY_DIR}/include")
aux_source_directory(src SRC_LIST_${PROJECT_NAME})
set(CMAKE_RUNTIME_OUTPUT_DIRECTORY
"${PROJECT_BINARY_DIR}/bin")
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY
"${PROJECT_BINARY_DIR}/bin/plugins")
set(CMAKE_ARCHIVE_OUTPUT_DIRECTORY
"${PROJECT_BINARY_DIR}/bin/plugins")
110
add_executable(${PROJECT_NAME} ${SRC_LIST_${PROJECT_NAME}})
set(${CONFIG_VAR_PREFIX}_USE_UNITY_BUILD TRUE CACHE BOOL "Use
unity build")
set(${CONFIG_VAR_PREFIX}_USE_PRECOMPILED_HEADER TRUE CACHE
BOOL "Use precompiled header")
set(${CONFIG_VAR_PREFIX}_BUILD_PLUGINS TRUE CACHE BOOL "Build
plugins")
enable_all_warnings(${PROJECT_NAME})
# Boost.Application needs dl library
target_link_libraries(${PROJECT_NAME} ${CMAKE_DL_LIBS})
target_link_libraries(${PROJECT_NAME} – lpthread – lboost_system – lrt –
lboost_chrono)
include_directories(BEFORE SYSTEM "${GOOGLE_BREAKPAD_DIR}/src")
setup_boost_settings()
find_package(Boost ${${CONFIG_VAR_PREFIX}_BOOST} COMPONENTS
date_time regex chrono atomic
system
thread
program_options
filesystem
log
log_setup
REQUIRED)
include_directories(BEFORE SYSTEM ${Boost_INCLUDE_DIRS})
target_link_libraries(${PROJECT_NAME} ${Boost_LIBRARIES})
set(EXECUTABLE_PATH
"${CMAKE_INSTALL_PREFIX}/bin/${PROJECT_NAME}")
111
set(CONFIG_FILE_PATH
"${CMAKE_INSTALL_PREFIX}/etc/${PROJECT_NAME}.conf")
set(PID_FILE_PATH "/var/run/${PROJECT_NAME}.pid")
set(LOG_DIR "/var/log/${PROJECT_NAME}")
install(TARGETS ${PROJECT_NAME}
RUNTIME DESTINATION bin
LIBRARY DESTINATION lib/RicGate
ARCHIVE DESTINATION lib/RicGate
)
configure_file("${CMAKE_SOURCE_DIR}/etc/RicGate.conf.in"
bin/etc/${PROJECT_NAME}.conf)
install(FILES "${PROJECT_BINARY_DIR}/etc/${PROJECT_NAME}.conf"
DESTINATION etc)
# configure_file("${CMAKE_SOURCE_DIR}/etc/doc/init.d.in"
etc/init.d/${PROJECT_NAME})
# install(PROGRAMS
#configure_file("${CMAKE_SOURCE_DIR}/include/config.hpp.in"
include/config.hpp)
add_subdirectory(plugins)
A.12 CMakeLists.txt – сборочный файл плагина
project(Mqtt)
cmake_minimum_required(VERSION 2.8)
message("CMAKE_MODULE_PATH=${CMAKE_MODULE_PATH}")
if(NOT DEFINED CMAKE_MODULE_PATH)
find_path(CMAKE_MODULE_PATH NAMES cmake
112
PATHS ${CMAKE_SOURCE_DIR}
PATH_SUFFIXES . .. ../.. NO_DEFAULT_PATH)
message("1CMAKE_MODULE_PATH=${CMAKE_MODULE_PATH}")
set(CMAKE_MODULE_PATH "${CMAKE_MODULE_PATH}/cmake" CACHE
INTERNAL "" FORCE)
message("2CMAKE_MODULE_PATH=${CMAKE_MODULE_PATH}")
endif()
# Include required CMake modules
include(utils)
is_target_enabled(${PROJECT_NAME} IS_ENABLED)
if(IS_ENABLED)
include_directories(../../include)
include_directories(.)
aux_source_directory(. SRC_LIST_${PROJECT_NAME})
add_library(${PROJECT_NAME}
MODULE
${SRC_LIST_${PROJECT_NAME}})
add_definitions(–
DMQTT_PARSERS_GRAMMAR_INSTANTIATE_HEADER_NAME="mqtt/parser
s/TemplateInstantiate.hpp")
enable_all_warnings(${PROJECT_NAME})
setup_boost_settings()
find_package(Boost
COMPONENTS
chrono
date_time
thread
REQUIRED)
include_directories(SYSTEM ${Boost_INCLUDE_DIRS})
target_link_libraries(${PROJECT_NAME} ${Boost_LIBRARIES})
install(TARGETS ${PROJECT_NAME}
RUNTIME DESTINATION bin
LIBRARY DESTINATION lib/ric_gate
system
log
113
ARCHIVE DESTINATION lib/ric_gate
)
endif()
A.13 InletMqtt.hpp – пример входного плагина (эхо– клиент)
#ifndef MQTT_PLUGIN_HPP
#define MQTT_PLUGIN_HPP
#include "log.hpp"
#include "api/PluginInlet.hpp"
#include "Dptree.hpp"
#include "Timer.hpp"
#include "jsonrpc.hpp"
#include <boost/enable_shared_from_this.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/array.hpp>
#include <boost/optional.hpp>
#include <iostream>
#include <queue>
namespace RicGate {
// Echo example translator.
class Mqtt
: public PluginApi::Inlet, public boost::enable_shared_from_this<Mqtt> {
public:
// Plugin constructor.
Mqtt();
// Plugin destructor.
~Mqtt();
// Plugin uid getter.
114
const char* uid() const
{ return "mqtt"; }
// Plugin name getter.
const char* name() const
{ return "MQTT"; }
// Plugin Protocol type
PluginApi::ProtocolType protocolType() const
{
return PluginApi::ProtocolType::TCP;
}
// Processing new connection
void startConnection(bool sessionMode);
// Processing after read some data
void processRead(std::size_t bytesTransferred);
// Action after write data to Inlet
void processWrite(bool isCommand);
// Processing command and call sendCommand();
bool
processCommand(boost::shared_ptr<dptree>
PluginApi::CommandCallback callback);
// Result callback for sendCommand() and usual packet
void processResult();
// Load configuration
void loadConfiguration(const boost::property_tree::ptree&);
private:
logging::logger log_;
boost::asio::streambuf streambuf;
std::vector<unsigned char> inBuffer_;
boost::array<unsigned char, 4096> readBuffer_;
};
data,
115
DECLARE_PLUGIN(Mqtt)
} // namespace RicGate
#endif // MQTT_HPP
A.14 InletMqtt.сpp – пример входного плагина (эхо– клиент)
#include "InletMqtt.hpp"
#include <boost/asio/buffers_iterator.hpp>
#include <boost/assign/list_of.hpp>
#include "DptreeJson.hpp"
namespace RicGate
{
Mqtt::Mqtt()
: log_(boost::log::keywords::channel = uid())
{
BOOST_LOG_SEV(log_, logging::debug) << name() << " created";
}
Mqtt::~Mqtt()
{
BOOST_LOG_SEV(log_, logging::debug) << name() << " destroyed";
boost::log::core::get()– >flush();
}
void Mqtt::startConnection(bool sessionMode)
{
api_– >readSome(boost::asio::buffer(readBuffer_));
}
void Mqtt::processRead(std::size_t bytesTransferred)
{
BOOST_LOG_SEV(log_, logging::debug)
116
<< "processRead(); " << "received " << bytesTransferred;
inBuffer_.assign(readBuffer_.begin(), readBuffer_.begin() + bytesTransferred);
api_– >readSome(boost::asio::buffer(readBuffer_));
}
void Mqtt::processWrite(bool isCommand)
{}
bool
Mqtt::processCommand(boost::shared_ptr<dptree>
PluginApi::CommandCallback callback)
{
BOOST_LOG_SEV(log_, logging::debug)
<< "processCommand()";
}
void Mqtt::processResult()
{}
void Mqtt::loadConfiguration(const boost::property_tree::ptree&)
{}
}// namespace RicGate
data,