README
¶
Streaming-Based Config
Streaming-Based Config is a helper library for accessing structured configurations based on event streaming/subscription.
Installation
To install the Streaming-Based-Config library, you can use go get:
go get github.com/Autodoc-Technology/streaming-based-config
Usage
Here's a basic example of how to use the Streaming-Based-Config library:
sub := sbc.NewSubscriber[SimpleConfig](
sbctransport.NewDemoTransport([]byte(`{"value": 1, "value2": "hello"}`), 3*time.Second),
sbc.DefaultKeyBuilder[SimpleConfig](),
)
confSubs, err := sub.Subscribe(ctx)
if err != nil {
log.Fatalf("Failed to subscribe: %v", err)
}
defer confSubs.Unsubscribe()
// get the current value
config := confSubs.Get()
log.Printf("Get Config: %+v", config)
// subscribe to the values stream
for val := range confSubs.GetUpdates() {
log.Printf("Config: %+v", val)
}
For more detailed examples, please refer to the example directory.
Development plan
- Basic code structure
- Unit tests for most of the code
- Nats transport implementation
- Consul transport implementation
- Unit tests for Subscriber/Subscriptions (with mocks)
- Kafka transport implementation
Contributing
We welcome contributions from the community. If you wish to contribute, please follow these steps:
- Fork the repository
- Create your feature branch (
git checkout -b feature/AmazingFeature) - Commit your changes (
git commit -m 'Add some AmazingFeature') - Push to the branch (
git push origin feature/AmazingFeature) - Open a pull request
Project Diagram
License
This project is Apache 2.0 licensed. For more information, please refer to the LICENSE file.
Contact
Project Link: https://github.com/Autodoc-Technology/streaming-based-config
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Encoder ¶
type Encoder interface {
// Encode encodes the input value v into a byte slice and returns it along with an error if the encoding fails.
Encode(v any) ([]byte, error)
// Decode decodes the provided byte slice into the target variable v of any type.
// Returns an error if the decoding process fails.
Decode(data []byte, v any) error
}
Encoder encodes the given value v into a byte slice. Returns the encoded data as a byte slice and an error if encoding fails.
type Holder ¶
type Holder[T any] struct { // contains filtered or unexported fields }
Holder represents a synchronized container that holds a value of any type.
func (*Holder[T]) GetValue ¶
func (e *Holder[T]) GetValue() T
GetValue returns the value of the envelope
type KeyBuilder ¶
type KeyBuilder[T any] interface { // BuildKey generates a string key based on the provided parameter of type T. BuildKey(t T) string }
KeyBuilder is an interface that defines a method to build a key based on a value of type T. The BuildKey method takes a value of type T as a parameter and returns a string key.
type KeyBuilderFunc ¶
KeyBuilderFunc is a type that represents a function that builds a key based on a value of type T. The function takes a value of type T as a parameter and returns a string key.
func (KeyBuilderFunc[T]) BuildKey ¶
func (k KeyBuilderFunc[T]) BuildKey(t T) string
BuildKey returns the string key generated by the KeyBuilderFunc for the given input value.
type Subscriber ¶
type Subscriber[T any] struct { // contains filtered or unexported fields }
Subscriber is a generic type that represents a subscriber for receiving updates from transport. It holds a reference to the transport, subscription, and subscriber options.
Use the Subscribe() method to start a new subscription by creating a new subscription instance and starting it. The subscription is started by calling the start() method of the subscription instance.
Use the Unsubscribe() method to stop the subscription by calling the stop() method of the subscription instance.
func NewSubscriber ¶
func NewSubscriber[T any](transport Transport, kb KeyBuilder[T], opts ...SubscriberOpt) *Subscriber[T]
NewSubscriber creates a new subscriber
func (*Subscriber[T]) Subscribe ¶
func (s *Subscriber[T]) Subscribe(ctx context.Context) (*Subscription[T], error)
Subscribe starts a new subscription by creating a new subscription instance and starting it.
The subscription is started by calling the start() method of the subscription instance. This method initializes internal variables, gets and decodes the initial value from the transport, and starts receiving updates from the transport in a separate goroutine.
Note that the context parameter is used to control the lifecycle of the subscription. When the context is canceled or done, the subscription's stop() method is called to stop the subscription and cancel the context.
If the subscription is not successfully started or the context is already done, the returned subscription instance will be nil.
type SubscriberOpt ¶
type SubscriberOpt func(*subscriberOpts)
SubscriberOpt is a function type used to configure a Subscriber instance. It modifies the options of the subscriberOpts struct.
func WithEncoder ¶
func WithEncoder(encoder Encoder) SubscriberOpt
WithEncoder sets the encoder option of a Subscriber instance.
type Subscription ¶
type Subscription[T any] struct { // contains filtered or unexported fields }
Subscription represents a Subscription to receive updates from a transport. It holds a reference to the transport, encoder, holder, and cancel function.
Use the Get() method to get the current value of the Subscription.
Use the GetUpdates() method to get a receive-only channel that sends updates of type T.
Use the start() method to start the Subscription and receive updates from the transport.
Use the stop() method to stop the Subscription and cancel the context.
Use the getAndDecode() method to get and decode the initial value from the transport.
Use the decode() method to decode the byte slice into type T.
func NewSubscription ¶
func NewSubscription[T any](transport Transport, encoder Encoder, kb KeyBuilder[T]) *Subscription[T]
NewSubscription creates a new subscription
func (*Subscription[T]) Get ¶
func (sub *Subscription[T]) Get() T
Get returns the current value of the subscription.
func (*Subscription[T]) GetUpdates ¶
func (sub *Subscription[T]) GetUpdates() <-chan T
GetUpdates returns a receive-only channel that sends updates of type T. The channel is closed when the specified context is done. It continuously sends the current value of the subscription holder to the channel. If the context is done, it stops sending updates and closes the channel.
func (*Subscription[T]) Unsubscribe ¶
func (sub *Subscription[T]) Unsubscribe()
Unsubscribe permanently stops the Subscription and cancel the context.
type Transport ¶
type Transport interface {
// Current retrieves the current value associated with the specified key using the given context.
// Returns the value as a byte slice and an error if retrieval fails.
Current(ctx context.Context, key string) ([]byte, error)
// Updates returns a channel that streams updates for the specified key using the given context.
// It stops and returns an error if the context is canceled.
Updates(ctx context.Context, key string) (<-chan []byte, error)
}
Transport defines an interface for accessing current data and subscribing to updates using a key and context.
Source Files
Directories