mirror of
https://github.com/harness/drone.git
synced 2025-05-01 13:11:27 +00:00
7abcacdd2f
This Commit adds: - stream package (provides different implementation of stream producers and consumers) + Redis -> will be used for any non-local deployments + InMemory -> a VERY BASIC implementation that is used for local execution - events package + GenericReporter -> responsible for reporting events, can be used to send any type of event and payload + GenericReader -> responsible for reading events from a stream, can be used to register handlers for any type of event and payload + ReaderFactory -> responsible for launching readers for any type of consumer group&name. - webhook package + The wire frame of the webhook package. - gitrpc/events package + defines event Reader/Reporter for events of category git
45 lines
1.2 KiB
Go
45 lines
1.2 KiB
Go
// Copyright 2022 Harness Inc. All rights reserved.
|
|
// Use of this source code is governed by the Polyform Free Trial License
|
|
// that can be found in the LICENSE.md file for this repository.
|
|
|
|
package stream
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
)
|
|
|
|
// MemoryProducer sends messages to streams of a MemoryBroker.
|
|
type MemoryProducer struct {
|
|
broker *MemoryBroker
|
|
// namespace specifies the namespace of the keys - any stream key will be prefixed with it
|
|
namespace string
|
|
}
|
|
|
|
func NewMemoryProducer(broker *MemoryBroker, namespace string) *MemoryProducer {
|
|
return &MemoryProducer{
|
|
broker: broker,
|
|
namespace: namespace,
|
|
}
|
|
}
|
|
|
|
// Send sends information to the Broker.
|
|
// Returns the message ID in case of success.
|
|
func (p *MemoryProducer) Send(ctx context.Context, streamID string, payload map[string]interface{}) (string, error) {
|
|
// ensure we transpose streamID using the key namespace
|
|
transposedStreamID := transposeStreamID(p.namespace, streamID)
|
|
|
|
msgID, err := p.broker.enqueue(
|
|
transposedStreamID,
|
|
message{
|
|
streamID: transposedStreamID,
|
|
values: payload,
|
|
})
|
|
if err != nil {
|
|
return "", fmt.Errorf("failed to write to stream '%s' (full stream '%s'). Error: %w",
|
|
streamID, transposedStreamID, err)
|
|
}
|
|
|
|
return msgID, nil
|
|
}
|