mirror of
https://github.com/hcengineering/platform.git
synced 2026-09-13 13:17:45 +02:00
remove to prepare for subtree
Signed-off-by: Denis Bykhov <bykhov.denis@gmail.com>
This commit is contained in:
Denis Bykhov
@@ -1,214 +0,0 @@
|
||||
// Copyright © 2025 Hardcore Engineering Inc.
|
||||
//
|
||||
// Licensed under the Eclipse Public License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License. You may
|
||||
// obtain a copy of the License at https://www.eclipse.org/legal/epl-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
//
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
// Package queue provides access to kafka queue
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/hcengineering/stream/internal/pkg/log"
|
||||
"github.com/segmentio/kafka-go"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// TopicTranscodeRequest is the name of the transcode requests topic.
|
||||
const TopicTranscodeRequest = "stream.transcode.request"
|
||||
|
||||
// TopicTranscodeResult is the name of the transcode results topic.
|
||||
const TopicTranscodeResult = "stream.transcode.result"
|
||||
|
||||
// TranscodeRequest represents transcode request.
|
||||
type TranscodeRequest struct {
|
||||
BlobID string
|
||||
WorkspaceUUID string
|
||||
ContentType string
|
||||
Source map[string]any
|
||||
}
|
||||
|
||||
// TranscodeResult represents transcode result.
|
||||
type TranscodeResult struct {
|
||||
BlobID string `json:"blobId"`
|
||||
WorkspaceUUID string `json:"workspaceUuid"`
|
||||
Source map[string]any `json:"source"`
|
||||
Playlist string `json:"playlist"`
|
||||
Thumbnail string `json:"thumbnail"`
|
||||
}
|
||||
|
||||
// ConsumerOptions represents options for the consumer
|
||||
type ConsumerOptions struct {
|
||||
Topic string
|
||||
Group string
|
||||
Config Config
|
||||
}
|
||||
|
||||
// Consumer provides a consumer interface to a Kafka queue
|
||||
type Consumer interface {
|
||||
Read(ctx context.Context) (kafka.Message, error)
|
||||
Fetch(ctx context.Context) (kafka.Message, error)
|
||||
Commit(ctx context.Context, msg kafka.Message) error
|
||||
Close() error
|
||||
}
|
||||
|
||||
// ProducerOptions represents options for the producer
|
||||
type ProducerOptions struct {
|
||||
Topic string
|
||||
Group string
|
||||
Config Config
|
||||
RetryCount int
|
||||
}
|
||||
|
||||
// Producer provides a producer interface to a Kafka queue
|
||||
type Producer interface {
|
||||
Send(ctx context.Context, workspaceID string, data any) error
|
||||
Close() error
|
||||
}
|
||||
|
||||
// TConsumer implements Consumer interface
|
||||
type TConsumer struct {
|
||||
topic string
|
||||
reader *kafka.Reader
|
||||
options ConsumerOptions
|
||||
}
|
||||
|
||||
// TProducer implements Producer interface
|
||||
type TProducer struct {
|
||||
topic string
|
||||
writer *kafka.Writer
|
||||
options ProducerOptions
|
||||
}
|
||||
|
||||
// Logger is kafka.Logger implementation
|
||||
type Logger struct {
|
||||
logger *zap.Logger
|
||||
}
|
||||
|
||||
var _ Producer = (*TProducer)(nil)
|
||||
var _ Consumer = (*TConsumer)(nil)
|
||||
var _ kafka.Logger = (*Logger)(nil)
|
||||
|
||||
// NewLogger creates new kafka.Logger
|
||||
func NewLogger(ctx context.Context) kafka.Logger {
|
||||
return &Logger{
|
||||
logger: log.FromContext(ctx),
|
||||
}
|
||||
}
|
||||
|
||||
// Printf logs a message
|
||||
func (l *Logger) Printf(msg string, args ...any) {
|
||||
l.logger.Sugar().Debugf(msg, args...)
|
||||
}
|
||||
|
||||
// NewConsumer creates a new transcoding request consumer
|
||||
func NewConsumer(ctx context.Context, options ConsumerOptions) Consumer {
|
||||
platformTopic := makeTopicID(options.Topic, options.Config)
|
||||
groupID := makeGroupID(options.Group, options.Topic, options.Config)
|
||||
|
||||
reader := kafka.NewReader(kafka.ReaderConfig{
|
||||
Brokers: options.Config.Brokers,
|
||||
GroupID: groupID,
|
||||
Topic: platformTopic,
|
||||
Logger: NewLogger(ctx),
|
||||
HeartbeatInterval: 1 * time.Second,
|
||||
SessionTimeout: 10 * time.Second,
|
||||
})
|
||||
|
||||
return &TConsumer{
|
||||
options: options,
|
||||
topic: platformTopic,
|
||||
reader: reader,
|
||||
}
|
||||
}
|
||||
|
||||
// Read reads (fetch and commit) a queue message.
|
||||
func (c *TConsumer) Read(ctx context.Context) (kafka.Message, error) {
|
||||
return c.reader.ReadMessage(ctx)
|
||||
}
|
||||
|
||||
// Fetch fetches a queue message.
|
||||
func (c *TConsumer) Fetch(ctx context.Context) (kafka.Message, error) {
|
||||
return c.reader.FetchMessage(ctx)
|
||||
}
|
||||
|
||||
// Commit commits the queue message.
|
||||
func (c *TConsumer) Commit(ctx context.Context, msg kafka.Message) error {
|
||||
return c.reader.CommitMessages(ctx, msg)
|
||||
}
|
||||
|
||||
// Close closes the consumer.
|
||||
func (c *TConsumer) Close() error {
|
||||
return c.reader.Close()
|
||||
}
|
||||
|
||||
// NewProducer creates a new transcoding result producer
|
||||
func NewProducer(ctx context.Context, options ProducerOptions) Producer {
|
||||
platformTopic := makeTopicID(options.Topic, options.Config)
|
||||
writer := &kafka.Writer{
|
||||
Addr: kafka.TCP(options.Config.Brokers...),
|
||||
Topic: platformTopic,
|
||||
AllowAutoTopicCreation: true,
|
||||
Logger: NewLogger(ctx),
|
||||
}
|
||||
return &TProducer{
|
||||
options: options,
|
||||
topic: platformTopic,
|
||||
writer: writer,
|
||||
}
|
||||
}
|
||||
|
||||
// Send sends a message to the queue topic
|
||||
func (p *TProducer) Send(ctx context.Context, workspaceID string, data any) error {
|
||||
if data == nil {
|
||||
return fmt.Errorf("event data is empty")
|
||||
}
|
||||
|
||||
value, err := json.Marshal(data)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for range p.options.RetryCount {
|
||||
err = p.writer.WriteMessages(
|
||||
ctx,
|
||||
kafka.Message{
|
||||
Key: []byte(workspaceID),
|
||||
Value: value,
|
||||
},
|
||||
)
|
||||
if err == nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// Close closes the producer.
|
||||
func (p *TProducer) Close() error {
|
||||
return p.writer.Close()
|
||||
}
|
||||
|
||||
func makeGroupID(groupID, topic string, config Config) string {
|
||||
return makeTopicID(topic, config) + "-" + groupID
|
||||
}
|
||||
|
||||
func makeTopicID(topic string, config Config) string {
|
||||
if config.Region != "" {
|
||||
return config.Region + "." + topic + config.Postfix
|
||||
}
|
||||
return topic + config.Postfix
|
||||
}
|
||||
Reference in New Issue
Block a user