Y3Codec Explainer

Проблема и мотивация

На рынке нет эффективного кодека для работы с постоянным соединением. Например, Protocol Buffers ориентирован на работу с данными в автономном режиме, в то время как пользователи сталкиваются с обработкой данных в реальном времени, которые генерируются с высокой частотой, а изменение структуры данных происходит редко.

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

Цели

• Быстрее, чем в реальном времени;
• Эффективное декодирование для обработки потока данных в реальном времени при постоянном соединении. Пользователям не нужно ждать получения полного пакета данных перед началом декодирования;
• Пользователям нужно только наблюдать за ключом, чтобы получить интересующие их данные.

Ключевые сценарии использования

• Приложения, чувствительные к задержкам;
• Обработка потока данных в реальном времени при постоянном соединении.

Предлагаемые решения

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

Y3 описывает объект как набор структур TLV. При декодировании пакетов он может заранее определить, является ли текущий T наблюдаемым ключом, и решить, следует ли перейти к следующему набору структур TLV напрямую, без ненужных операций декодирования для ненаблюдаемых пакетов. Таким образом, Y3 может повысить эффективность декодирования и использование ресурсов.

Основные интерфейсы включают:

• Marshal — сериализует данные пользователя в соответствии с правилами кодирования Y3;
• Subscribe — наблюдает за ключом, указанным пользователем;
• OnObserve — запускает функцию обратного вызова, когда ключ наблюдается Y3.

Примеры

1. Источник данных представляет собой пакет JSON (включая интересующие и неинтересные данные). Эти данные необходимо закодировать с помощью Y3 и передать получателю по потоковой передаче, например, yomo-flow. Получатель наблюдает интересующие данные и обрабатывает их.

Кодирование данных

type SourceData struct {
   Name  string      `y3:"0x10"`
   Noise float32     `y3:"0x11"`
   Therm Thermometer `y3:"0x12"`
}

type Thermometer struct {
   Temperature float32 `y3:"0x13"`
   Humidity    float32 `y3:"0x14"`
}

func main() {
	input := SourceData{
		Name:  "yomo",
		Noise: float32(456),
		Therm: Thermometer{Temperature: float32(30), Humidity: float32(40)},
	}
  // кодируем данные в формате Y3-Codec
	codec := y3.NewCodec(0x20)
	inputBuf, _ := codec.Marshal(input)
	fmt.Printf("inputBuf=%#v\n", inputBuf)
}

Наблюдение за указанным ключом и декодирование

func main() {
	// определяем функцию обратного вызова для обработки наблюдаемых данных
	callback := func(v []byte) (interface{}, error) {
		return y3.ToFloat32(v)
	}
	// создаём интерфейс Observable
	source := y3.FromStream(bytes.NewReader(inputBuf))
  // подписываемся на наблюдаемый ключ и устанавливаем функцию обратного вызова
	consumer := source.Subscribe(0x11).OnObserve(callback)
	// проверка данных после их обработки
	for c := range consumer {
		fmt.Printf("observed value=%v, type=%v\n", c, reflect.ValueOf(c).Kind())
	}
}

Внимание

Диапазон ключа, на который подписываются:

• Определяемый пользователем ключ для подписки: 0x10 ~ 0x3f;
• Зарезервированный системой: 0x01 ~ 0x0f.