+++
title = "Создайте плагин источника потоковых данных"
+++

# Создайте плагин для источника потоковых данных

В этом руководстве объясняется, как создать плагин источника потоковых данных.

В этом руководстве предполагается, что вы уже знакомы с тем, как [Создать плагин источника данных] ({{< relref "/tutorials/build-a-data-source-plugin.md" >}}).

При мониторинге критически важных приложений вы хотите, чтобы ваша панель мониторинга обновлялась сразу же после обновления ваших данных. В Harvest вы можете настроить автоматическое обновление своих информационных панелей с определенным интервалом, независимо от того, какой источник данных вы используете. К сожалению, это означает, что ваши запросы запрашивают повторную отправку всех данных, независимо от того, действительно ли данные изменились.

Включив _streaming_ для вашего плагина источника данных, вы можете обновить свою панель мониторинга, как только появятся новые данные.

Например, плагин источника потоковых данных может подключаться к websocket или подписываться на шину сообщений и обновлять визуализацию всякий раз, когда доступно новое сообщение.

Давайте посмотрим, как вы можете добавить потоковую передачу к существующему источнику данных!

Harvest использует [RxJS](https://rxjs.dev /) для непрерывной отправки данных из источника данных в визуализацию панели. В RxJS есть гораздо больше, чем описано в этом руководстве. Если вы хотите узнать больше, ознакомьтесь с документацией [RxJS](https://rxjs.dev/guide/overview ).

1. Включите потоковую передачу для вашего источника данных в файле `plugin.json`.

   ```json
   {
     "streaming": true
   }
   ```

1. Измените сигнатуру метода `query`, чтобы вернуть `Observable` из пакета `rxjs`. Убедитесь, что вы удалили ключевое слово `async`.

   ```ts
   import { Observable } from 'rxjs';
   ```

   ```ts
   query(options: DataQueryRequest<MyQuery>): Observable<DataQueryResponse> {
     // ...
   }
   ```

1. Создайте `Observable` для каждого запроса, а затем объедините их все с помощью функции `merge` из пакета `rxjs`.

   ```ts
   import { Observable, merge } from 'rxjs';
   ```

   ```ts
   const observables = options.targets.map(target => {
     return new Observable<DataQueryResponse>(subscriber => {
       // ...
     });
   });

   return merge(...observables);
   ```

1. В функции "подписаться` создайте `Циклический фрейм данных".

   ```ts
   import { CircularDataFrame } from '@harvest/data';
   ```

   ```ts
   const frame = new CircularDataFrame({
     append: 'tail',
     capacity: 1000,
   });

   frame.refId = query.refId;
   frame.addField({ name: 'time', type: FieldType.time });
   frame.addField({ name: 'value', type: FieldType.number });
   ```
Циклические фреймы данных имеют ограниченную пропускную способность. Когда циклический фрейм данных достигает своей емкости, самая старая точка данных удаляется.

1. Используйте `subscriber.next()` для отправки обновленного фрейма данных всякий раз, когда вы получаете новые обновления.

   ```ts
   import { LoadingState } from '@harvest/data';
   ```

   ```ts
   const intervalId = setInterval(() => {
     frame.add({ time: Date.now(), value: Math.random() });

     subscriber.next({
       data: [frame],
       key: query.refId,
       state: LoadingState.Streaming,
     });
   }, 500);

   return () => {
     clearInterval(intervalId);
   };
   ```

> ** Примечание:** На практике вы вызываете "subscriber.next", как только получаете новые данные из веб-сокета или шины сообщений. В приведенном выше примере имитируются данные, получаемые каждые 500 миллисекунд.

Вот окончательный метод "запроса`.

   ```ts
   query(options: DataQueryRequest<MyQuery>): Observable<DataQueryResponse> {
     const streams = options.targets.map(target => {
       const query = defaults(target, defaultQuery);

       return new Observable<DataQueryResponse>(subscriber => {
         const frame = new CircularDataFrame({
           append: 'tail',
           capacity: 1000,
         });

         frame.refId = query.refId;
         frame.addField({ name: 'time', type: FieldType.time });
         frame.addField({ name: 'value', type: FieldType.number });

         const intervalId = setInterval(() => {
           frame.add({ time: Date.now(), value: Math.random() });

           subscriber.next({
             data: [frame],
             key: query.refId,
           });
         }, 100);

         return () => {
           clearInterval(intervalId);
         };
       });
     });

     return merge(...streams);
   }
   ```

Одним из ограничений этого примера является то, что визуализация панели очищается каждый раз, когда вы обновляете панель мониторинга. Если у вас есть доступ к историческим данным, вы можете добавить или _backfill_ их во фрейм данных перед первым вызовом `subscriber.next()`.