107 lines
3.5 KiB
Go
107 lines
3.5 KiB
Go
/*
|
|
Copyright 2021 The Dapr Authors
|
|
Licensed under the Apache 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
|
|
http://www.apache.org/licenses/LICENSE-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 eventhubs
|
|
|
|
import (
|
|
"context"
|
|
"reflect"
|
|
|
|
"github.com/Azure/azure-sdk-for-go/sdk/messaging/azeventhubs"
|
|
|
|
"github.com/dapr/components-contrib/bindings"
|
|
impl "github.com/dapr/components-contrib/common/component/azure/eventhubs"
|
|
contribMetadata "github.com/dapr/components-contrib/metadata"
|
|
"github.com/dapr/kit/logger"
|
|
"github.com/dapr/kit/ptr"
|
|
)
|
|
|
|
// AzureEventHubs allows sending/receiving Azure Event Hubs events.
|
|
type AzureEventHubs struct {
|
|
*impl.AzureEventHubs
|
|
}
|
|
|
|
// NewAzureEventHubs returns a new Azure Event hubs instance.
|
|
func NewAzureEventHubs(logger logger.Logger) bindings.InputOutputBinding {
|
|
return &AzureEventHubs{
|
|
AzureEventHubs: impl.NewAzureEventHubs(logger, true),
|
|
}
|
|
}
|
|
|
|
// Init performs metadata init.
|
|
func (a *AzureEventHubs) Init(_ context.Context, metadata bindings.Metadata) error {
|
|
return a.AzureEventHubs.Init(metadata.Properties)
|
|
}
|
|
|
|
func (a *AzureEventHubs) Operations() []bindings.OperationKind {
|
|
return []bindings.OperationKind{
|
|
bindings.CreateOperation,
|
|
}
|
|
}
|
|
|
|
// Write posts an event hubs message.
|
|
func (a *AzureEventHubs) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
|
|
// Get the partition key and content type
|
|
batchOpts := &azeventhubs.EventDataBatchOptions{}
|
|
if pk := req.Metadata["partitionKey"]; pk != "" {
|
|
batchOpts.PartitionKey = &pk
|
|
}
|
|
var contentType *string
|
|
if ct := req.Metadata["contentType"]; ct != "" {
|
|
contentType = ptr.Of(req.Metadata["contentType"])
|
|
}
|
|
|
|
// Publish the message
|
|
messages := []*azeventhubs.EventData{
|
|
{
|
|
Body: req.Data,
|
|
ContentType: contentType,
|
|
},
|
|
}
|
|
|
|
// Publish the message
|
|
err := a.AzureEventHubs.Publish(ctx, a.AzureEventHubs.EventHubName(), messages, batchOpts)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return nil, nil
|
|
}
|
|
|
|
// Read gets messages from Event Hubs in a non-blocking way.
|
|
func (a *AzureEventHubs) Read(ctx context.Context, handler bindings.Handler) error {
|
|
// Start the subscription
|
|
// This is non-blocking
|
|
topic := a.AzureEventHubs.EventHubName()
|
|
bindingsHandler := a.AzureEventHubs.GetBindingsHandlerFunc(topic, false, handler)
|
|
// Setting `maxBulkSubCount` to 1 as bindings are not supported for bulk subscriptions
|
|
// Setting `CheckPointFrequencyPerPartition` to default value of 1
|
|
return a.AzureEventHubs.Subscribe(ctx, impl.SubscribeConfig{
|
|
Topic: topic,
|
|
MaxBulkSubCount: 1,
|
|
MaxBulkSubAwaitDurationMs: impl.DefaultMaxBulkSubAwaitDurationMs,
|
|
CheckPointFrequencyPerPartition: impl.DefaultCheckpointFrequencyPerPartition,
|
|
Handler: bindingsHandler,
|
|
})
|
|
}
|
|
|
|
func (a *AzureEventHubs) Close() error {
|
|
return a.AzureEventHubs.Close()
|
|
}
|
|
|
|
// GetComponentMetadata returns the metadata of the component.
|
|
func (a *AzureEventHubs) GetComponentMetadata() (metadataInfo contribMetadata.MetadataMap) {
|
|
metadataStruct := impl.AzureEventHubsMetadata{}
|
|
contribMetadata.GetMetadataInfoFromStructType(reflect.TypeOf(metadataStruct), &metadataInfo, contribMetadata.BindingType)
|
|
return
|
|
}
|