mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Stephan BehnkeandGitHub
9fd2383c0c
## What changed? Removes the `encodingType` from the `Serializer` interface. ## Why? The parameter is superfluous since it always set to `enumspb.ENCODING_TYPE_PROTO3`. There are no plans to move to a different encoding scheme. An encoding parameter for existing/new methods can be added (back) as needed. The next steps will be: 1. a codec for the serializer to allow customizing the encoding globally 1. an env variable to customize the encoding to be JSON (for debugging/developing purposes) ## How did you test it? - [ ] built - [ ] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s)
139 lines
3.7 KiB
Go
139 lines
3.7 KiB
Go
package persistence
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"go.temporal.io/api/serviceerror"
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/persistence/serialization"
|
|
)
|
|
|
|
var (
|
|
ErrNexusTableVersionConflict = serviceerror.NewFailedPrecondition("nexus endpoints table version mismatch")
|
|
ErrNexusEndpointVersionConflict = serviceerror.NewFailedPrecondition("nexus endpoint version mismatch")
|
|
ErrNegativeListNexusEndpointsPageSize = &InvalidPersistenceRequestError{
|
|
Msg: "received negative page size for listing Nexus endpoints",
|
|
}
|
|
)
|
|
|
|
type (
|
|
nexusEndpointManagerImpl struct {
|
|
persistence NexusEndpointStore
|
|
serializer serialization.Serializer
|
|
logger log.Logger
|
|
}
|
|
)
|
|
|
|
func NewNexusEndpointManager(
|
|
persistence NexusEndpointStore,
|
|
serializer serialization.Serializer,
|
|
logger log.Logger,
|
|
) NexusEndpointManager {
|
|
return &nexusEndpointManagerImpl{
|
|
persistence: persistence,
|
|
serializer: serializer,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
func (m *nexusEndpointManagerImpl) GetName() string {
|
|
return m.persistence.GetName()
|
|
}
|
|
|
|
func (m *nexusEndpointManagerImpl) Close() {
|
|
m.persistence.Close()
|
|
}
|
|
|
|
func (m *nexusEndpointManagerImpl) GetNexusEndpoint(
|
|
ctx context.Context,
|
|
request *GetNexusEndpointRequest,
|
|
) (*persistencespb.NexusEndpointEntry, error) {
|
|
internalEndpoint, err := m.persistence.GetNexusEndpoint(ctx, request)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
endpoint, err := m.serializer.NexusEndpointFromBlob(internalEndpoint.Data)
|
|
if err != nil {
|
|
m.logger.Error(fmt.Sprintf("error deserializing nexus endpoint with ID:%v", internalEndpoint.ID), tag.Error(err))
|
|
return nil, err
|
|
}
|
|
|
|
return &persistencespb.NexusEndpointEntry{
|
|
Id: internalEndpoint.ID,
|
|
Version: internalEndpoint.Version,
|
|
Endpoint: endpoint,
|
|
}, nil
|
|
}
|
|
|
|
func (m *nexusEndpointManagerImpl) ListNexusEndpoints(
|
|
ctx context.Context,
|
|
request *ListNexusEndpointsRequest,
|
|
) (*ListNexusEndpointsResponse, error) {
|
|
if request.PageSize < 0 {
|
|
return nil, ErrNegativeListNexusEndpointsPageSize
|
|
}
|
|
|
|
result := &ListNexusEndpointsResponse{}
|
|
|
|
resp, err := m.persistence.ListNexusEndpoints(ctx, request)
|
|
if resp != nil {
|
|
result.TableVersion = resp.TableVersion
|
|
}
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
|
|
entries := make([]*persistencespb.NexusEndpointEntry, len(resp.Endpoints))
|
|
for i, entry := range resp.Endpoints {
|
|
endpoint, err := m.serializer.NexusEndpointFromBlob(entry.Data)
|
|
if err != nil {
|
|
m.logger.Error(fmt.Sprintf("error deserializing nexus endpoint with ID: %v", entry.ID), tag.Error(err))
|
|
return nil, err
|
|
}
|
|
entries[i] = &persistencespb.NexusEndpointEntry{
|
|
Id: entry.ID,
|
|
Version: entry.Version,
|
|
Endpoint: endpoint,
|
|
}
|
|
}
|
|
|
|
result.NextPageToken = resp.NextPageToken
|
|
result.Entries = entries
|
|
return result, nil
|
|
}
|
|
|
|
func (m *nexusEndpointManagerImpl) CreateOrUpdateNexusEndpoint(
|
|
ctx context.Context,
|
|
request *CreateOrUpdateNexusEndpointRequest,
|
|
) (*CreateOrUpdateNexusEndpointResponse, error) {
|
|
blob, err := m.serializer.NexusEndpointToBlob(request.Entry.Endpoint)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = m.persistence.CreateOrUpdateNexusEndpoint(ctx, &InternalCreateOrUpdateNexusEndpointRequest{
|
|
LastKnownTableVersion: request.LastKnownTableVersion,
|
|
Endpoint: InternalNexusEndpoint{
|
|
ID: request.Entry.Id,
|
|
Version: request.Entry.Version,
|
|
Data: blob,
|
|
},
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &CreateOrUpdateNexusEndpointResponse{Version: request.Entry.Version + 1}, nil
|
|
}
|
|
|
|
func (m *nexusEndpointManagerImpl) DeleteNexusEndpoint(
|
|
ctx context.Context,
|
|
request *DeleteNexusEndpointRequest,
|
|
) error {
|
|
return m.persistence.DeleteNexusEndpoint(ctx, request)
|
|
}
|