Skip to content

Commit afa721d

Browse files
fengmk2fengmk2
committed
First impl for the State API
See https://mosn.io/layotto/#/zh/building_blocks/state/reference
1 parent 12b8ca7 commit afa721d

File tree

7 files changed

+311
-106
lines changed

7 files changed

+311
-106
lines changed

‎sdk/js-sdk/README.md

+2-1
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,8 @@ await client.state.save(storeName, { key, value });
1919
console.log('saveState success, key: %j, value: %j', key, value);
2020

2121
const resValue = await client.state.get(storeName, key);
22-
console.log('getState success, key: %j, value: %j, toString: %j', key, resValue, Buffer.from(resValue).toString('utf8'));
22+
console.log('getState success, key: %j, value: %j, toString: %j',
23+
key, resValue, Buffer.from(resValue).toString('utf8'));
2324
```
2425

2526
## Development

‎sdk/js-sdk/demo/state.ts

+7-5
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,16 @@ async function main() {
1111

1212
await client.state.save(storeName, [
1313
{ key, value },
14-
]);
14+
], { traceid: 'mock-tracerid-123123' });
1515
console.log('saveState success, key: %j, value: %j', key, value);
1616

17-
await client.state.save(storeName, { key, value });
18-
console.log('saveState success, key: %j, value: %j', key, value);
17+
// await client.state.save(storeName, { key, value });
18+
// console.log('saveState success, key: %j, value: %j', key, value);
1919

20-
const resValue = await client.state.get(storeName, key);
21-
console.log('getState success, key: %j, value: %j, toString: %j', key, resValue, Buffer.from(resValue).toString('utf8'));
20+
const state = await client.state.get(storeName, key);
21+
assert(state);
22+
console.log('getState success, key: %j, value: %j, toString: %j',
23+
key, state.value, Buffer.from(state.value).toString('utf8'));
2224
}
2325

2426
main();

‎sdk/js-sdk/src/API.ts

+15-1
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,22 @@
1+
import { Metadata } from '@grpc/grpc-js';
12
import { RuntimeClient } from '../proto/runtime_grpc_pb';
23

3-
export default class API {
4+
// gRPC meta data
5+
export type RequestMetadata = {
6+
[key: string]: string;
7+
};
8+
9+
export class API {
410
readonly runtime: RuntimeClient;
511
constructor(runtime: RuntimeClient) {
612
this.runtime = runtime;
713
}
14+
15+
createMetadata(meta?: RequestMetadata) {
16+
const metadata = new Metadata();
17+
for (const key in meta) {
18+
metadata.add(key, meta[key]);
19+
}
20+
return metadata;
21+
}
822
}

‎sdk/js-sdk/src/Hello.ts

+3-3
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,18 @@
11
import {
22
SayHelloRequest
33
} from '../proto/runtime_pb';
4-
import API from './API';
4+
import { API, RequestMetadata } from './API';
55

66
export default class Hello extends API {
7-
async sayHello(serviceName = 'helloworld', name = ''): Promise<string> {
7+
async sayHello(serviceName = 'helloworld', name = '', meta?: RequestMetadata): Promise<string> {
88
const req = new SayHelloRequest();
99
req.setServiceName(serviceName);
1010
if (name) {
1111
req.setName(name);
1212
}
1313

1414
return new Promise((resolve, reject) => {
15-
this.runtime.sayHello(req, (err, res) => {
15+
this.runtime.sayHello(req, this.createMetadata(meta), (err, res) => {
1616
if (err) return reject(err);
1717
resolve(res.getHello());
1818
});

‎sdk/js-sdk/src/State.ts

+124-60
Original file line numberDiff line numberDiff line change
@@ -1,109 +1,85 @@
11
import {
22
SaveStateRequest,
33
StateItem as StateItemPB,
4-
Etag as ETagTB,
4+
Etag,
55
StateOptions as StateOptionsPB,
66
GetStateRequest,
77
GetBulkStateRequest,
8+
DeleteStateRequest,
9+
DeleteBulkStateRequest,
10+
ExecuteStateTransactionRequest,
11+
TransactionalStateOperation,
812
} from '../proto/runtime_pb';
9-
import API from './API';
10-
11-
export type StateConcurrency = StateOptionsPB.StateConcurrency;
12-
export type StateConsistency = StateOptionsPB.StateConsistency;
13-
14-
type StateOptions = {
15-
concurrency: StateConcurrency;
16-
consistency: StateConsistency;
17-
};
18-
19-
type Metadata = {
20-
[key: string]: string;
21-
};
22-
23-
type StateItem = {
24-
key: string;
25-
value: Uint8Array | string;
26-
etag?: string;
27-
metadata?: Metadata;
28-
options?: StateOptions;
29-
}
30-
31-
type BulkResponseStateItem = {
32-
key: string;
33-
value: Uint8Array;
34-
etag: string;
35-
}
13+
import { API, RequestMetadata } from './API';
14+
import {
15+
DeleteStateItem,
16+
ResponseStateItem,
17+
StateItem,
18+
StateOperation,
19+
StateOptions,
20+
} from './types/State';
3621

3722
export default class State extends API {
3823
// Saves an array of state objects
39-
async save(storeName: string, states: StateItem[] | StateItem): Promise<void> {
40-
const stateList: StateItemPB[] = [];
24+
async save(storeName: string, states: StateItem[] | StateItem, meta?: RequestMetadata): Promise<void> {
4125
if (!Array.isArray(states)) {
4226
states = [states];
4327
}
44-
for (const item of states) {
45-
const stateItem = new StateItemPB();
46-
stateItem.setKey(item.key);
47-
if (typeof item.value === 'string') {
48-
stateItem.setValue(Buffer.from(item.value, 'utf8'));
49-
} else {
50-
stateItem.setValue(item.value);
51-
}
52-
if (item.etag !== undefined) {
53-
const etag = new ETagTB();
54-
etag.setValue(item.etag);
55-
stateItem.setEtag(etag);
56-
}
57-
if (item.options !== undefined) {
58-
const options = new StateOptionsPB();
59-
options.setConcurrency(item.options.concurrency);
60-
}
61-
stateList.push(stateItem);
62-
}
63-
28+
const stateList = this.createStateItemPBList(states);
6429
const req = new SaveStateRequest();
6530
req.setStoreName(storeName);
6631
req.setStatesList(stateList);
6732

6833
return new Promise((resolve, reject) => {
69-
this.runtime.saveState(req, (err) => {
34+
this.runtime.saveState(req, this.createMetadata(meta), (err) => {
7035
if (err) return reject(err);
7136
resolve();
7237
});
7338
});
7439
}
7540

7641
// Gets the state for a specific key
77-
async get(storeName: string, key: string): Promise<Uint8Array> {
42+
async get(storeName: string, key: string, meta?: RequestMetadata): Promise<ResponseStateItem | null> {
7843
const req = new GetStateRequest();
7944
req.setStoreName(storeName);
8045
req.setKey(key);
8146

8247
return new Promise((resolve, reject) => {
83-
this.runtime.getState(req, (err, res) => {
48+
this.runtime.getState(req, this.createMetadata(meta), (err, res) => {
8449
if (err) return reject(err);
85-
resolve(res.getData_asU8());
50+
if (this.isEmpty(res)) {
51+
return resolve(null);
52+
}
53+
resolve({
54+
key,
55+
value: res.getData_asU8(),
56+
etag: res.getEtag(),
57+
});
8658
});
8759
});
8860
}
8961

9062
// Gets a bulk of state items for a list of keys
91-
async getBulk(storeName: string, keys: string[], parallelism = 10): Promise<BulkResponseStateItem[]> {
63+
async getBulk(storeName: string, keys: string[], parallelism = 10, meta?: RequestMetadata): Promise<ResponseStateItem[]> {
9264
const req = new GetBulkStateRequest();
9365
req.setStoreName(storeName);
9466
req.setKeysList(keys);
9567
req.setParallelism(parallelism);
9668

9769
return new Promise((resolve, reject) => {
98-
this.runtime.getBulkState(req, (err, res) => {
70+
this.runtime.getBulkState(req, this.createMetadata(meta), (err, res) => {
9971
if (err) return reject(err);
100-
const states: BulkResponseStateItem[] = [];
72+
const states: ResponseStateItem[] = [];
10173
const itemsList = res.getItemsList();
10274
for (const item of itemsList) {
75+
if (this.isEmpty(item)) {
76+
continue;
77+
}
10378
states.push({
10479
key: item.getKey(),
10580
value: item.getData_asU8(),
10681
etag: item.getEtag(),
82+
// metadata: item.getMetadataMap(),
10783
});
10884
}
10985
resolve(states);
@@ -112,12 +88,100 @@ export default class State extends API {
11288
}
11389

11490
// Deletes the state for a specific key
115-
async delete() {
91+
async delete(storeName: string, key: string, etag = '', options?: StateOptions, meta?: RequestMetadata): Promise<void> {
92+
const req = new DeleteStateRequest();
93+
req.setStoreName(storeName);
94+
req.setKey(key);
95+
if (etag) {
96+
const etagInstance = new Etag();
97+
etagInstance.setValue(etag);
98+
req.setEtag(etagInstance);
99+
}
100+
if (options) {
101+
const optionsInstance = new StateOptionsPB();
102+
optionsInstance.setConcurrency(options.concurrency);
103+
optionsInstance.setConsistency(options.consistency);
104+
req.setOptions(optionsInstance);
105+
}
116106

107+
return new Promise((resolve, reject) => {
108+
this.runtime.deleteState(req, this.createMetadata(meta), (err) => {
109+
if (err) return reject(err);
110+
resolve();
111+
});
112+
});
117113
}
118114

119115
// Deletes a bulk of state items for a list of keys
120-
async deleteBulk() {}
116+
async deleteBulk(storeName: string, states: DeleteStateItem[], meta?: RequestMetadata): Promise<void> {
117+
const req = new DeleteBulkStateRequest();
118+
req.setStoreName(storeName);
119+
const stateList = this.createStateItemPBList(states);
120+
req.setStatesList(stateList);
121+
122+
return new Promise((resolve, reject) => {
123+
this.runtime.deleteBulkState(req, this.createMetadata(meta), (err) => {
124+
if (err) return reject(err);
125+
resolve();
126+
});
127+
});
128+
}
129+
130+
// Executes transactions for a specified store
131+
async executeTransaction(storeName: string, operations: StateOperation[], meta?: RequestMetadata): Promise<void> {
132+
const req = new ExecuteStateTransactionRequest();
133+
req.setStorename(storeName);
134+
const operationsList: TransactionalStateOperation[] = [];
135+
for (const operation of operations) {
136+
const ops = new TransactionalStateOperation();
137+
ops.setOperationtype(operation.operationType);
138+
const stateItem = this.createStateItemPB(operation.request);
139+
ops.setRequest(stateItem);
140+
operationsList.push(ops);
141+
}
142+
req.setOperationsList(operationsList);
143+
144+
return new Promise((resolve, reject) => {
145+
this.runtime.executeStateTransaction(req, this.createMetadata(meta), (err, _res) => {
146+
if (err) return reject(err);
147+
resolve();
148+
});
149+
});
150+
}
121151

122-
// async transaction() {}
152+
private createStateItemPB(item: StateItem | DeleteStateItem): StateItemPB {
153+
const stateItem = new StateItemPB();
154+
stateItem.setKey(item.key);
155+
if ('value' in item) {
156+
if (typeof item.value === 'string') {
157+
stateItem.setValue(Buffer.from(item.value, 'utf8'));
158+
} else {
159+
stateItem.setValue(item.value);
160+
}
161+
}
162+
if (item.etag !== undefined) {
163+
const etag = new Etag();
164+
etag.setValue(item.etag);
165+
stateItem.setEtag(etag);
166+
}
167+
if (item.options !== undefined) {
168+
const options = new StateOptionsPB();
169+
options.setConcurrency(item.options.concurrency);
170+
options.setConsistency(item.options.consistency);
171+
stateItem.setOptions(options);
172+
}
173+
return stateItem;
174+
}
175+
176+
private createStateItemPBList(items: StateItem[] | DeleteStateItem[]): StateItemPB[] {
177+
const list: StateItemPB[] = [];
178+
for (const item of items) {
179+
list.push(this.createStateItemPB(item));
180+
}
181+
return list;
182+
}
183+
184+
private isEmpty(obj: { getEtag(): string }) {
185+
return obj.getEtag() === '';
186+
}
123187
}

‎sdk/js-sdk/src/types/State.ts

+42
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
import {
2+
StateOptions as StateOptionsPB,
3+
} from '../../proto/runtime_pb';
4+
5+
export type StateConcurrency = StateOptionsPB.StateConcurrency;
6+
export type StateConsistency = StateOptionsPB.StateConsistency;
7+
8+
export type StateOptions = {
9+
concurrency: StateConcurrency;
10+
consistency: StateConsistency;
11+
};
12+
13+
export type StateItem = {
14+
key: string;
15+
value: Uint8Array | string;
16+
etag?: string;
17+
options?: StateOptions;
18+
}
19+
20+
export type DeleteStateItem = {
21+
key: string;
22+
etag?: string;
23+
options?: StateOptions;
24+
};
25+
26+
export type ResponseStateItem = {
27+
key: string;
28+
value: Uint8Array;
29+
etag: string;
30+
// metadata;
31+
}
32+
33+
export enum StateOperationType {
34+
Upsert = 'upsert',
35+
Delete = 'delete',
36+
}
37+
38+
export type StateOperation = {
39+
operationType: StateOperationType;
40+
request: StateItem | DeleteStateItem;
41+
};
42+

0 commit comments

Comments
 (0)