50
session gocql.Session,
51
logger log.Logger,
53
>
return &ShardStore{
54
>
ClusterName: clusterName,
55
>
Session: session,
56
>
Logger: logger,
57
>
}
58
>
}
59
60
func (d *ShardStore) GetOrCreateShard(
61
ctx context.Context,
62
request *p.InternalGetOrCreateShardRequest,
64
>
query := d.Session.Query(templateGetShardQuery,
65
>
request.ShardID,
66
>
rowTypeShard,
67
>
rowTypeShardNamespaceID,
68
>
rowTypeShardWorkflowID,
69
>
rowTypeShardRunID,
70
>
defaultVisibilityTimestamp,
71
>
rowTypeShardTaskID,
72
>
).WithContext(ctx)
73
>
74
>
var data []byte
75
>
var encoding string
76
>
err := query.Scan(&data, &encoding)
77
>
if err == nil {
78
return &p.InternalGetOrCreateShardResponse{
79
ShardInfo: p.NewDataBlob(data, encoding),
80
}, nil
82
return nil, gocql.ConvertError("GetOrCreateShard", err)
83
}
84
85
// shard was not found and we should create it
87
>
if err != nil {
88
return nil, err
89
}
90
92
>
request.ShardID,
93
>
rowTypeShard,
94
>
rowTypeShardNamespaceID,
95
>
rowTypeShardWorkflowID,
96
>
rowTypeShardRunID,
97
>
defaultVisibilityTimestamp,
98
>
rowTypeShardTaskID,
99
>
shardInfo.Data,
100
>
shardInfo.EncodingType.String(),
101
>
rangeID,
102
>
).WithContext(ctx)
103
>
104
>
previous := make(map[string]any)
105
>
applied, err := query.MapScanCAS(previous)
106
>
if err != nil {
107
return nil, gocql.ConvertError("GetOrCreateShard", err)
108
}
110
// conflict, try again
111
request.CreateShardInfo = nil // prevent loop
112
return d.GetOrCreateShard(ctx, request)
113
}
115
>
ShardInfo: shardInfo,
116
>
}, nil
117
}
118