Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit fc00af2

Browse files
nullndrroggervalf
nullndr
authored and
roggervalf
committedSep 13, 2024
feat: add Serialize type
1 parent f6546fb commit fc00af2

File tree

3 files changed

+174
-27
lines changed

3 files changed

+174
-27
lines changed
 

‎src/classes/worker.ts

+27-27
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ import {
1515
RedisClient,
1616
WorkerOptions,
1717
} from '../interfaces';
18-
import { MinimalQueue } from '../types';
18+
import { MinimalQueue, Serialize } from '../types';
1919
import {
2020
delay,
2121
DELAY_TIME_1,
@@ -55,7 +55,7 @@ export interface WorkerListener<
5555
*
5656
* This event is triggered when a job enters the 'active' state.
5757
*/
58-
active: (job: Job<DataType, ResultType, NameType>, prev: string) => void;
58+
active: (job: Job<Serialize<DataType>, ResultType, NameType>, prev: string) => void;
5959

6060
/**
6161
* Listen to 'closing' event.
@@ -77,7 +77,7 @@ export interface WorkerListener<
7777
* This event is triggered when a job has successfully completed.
7878
*/
7979
completed: (
80-
job: Job<DataType, ResultType, NameType>,
80+
job: Job<Serialize<DataType>, ResultType, NameType>,
8181
result: ResultType,
8282
prev: string,
8383
) => void;
@@ -106,7 +106,7 @@ export interface WorkerListener<
106106
* reaches the stalled limit and it is deleted by the removeOnFail option.
107107
*/
108108
failed: (
109-
job: Job<DataType, ResultType, NameType> | undefined,
109+
job: Job<Serialize<DataType>, ResultType, NameType> | undefined,
110110
error: Error,
111111
prev: string,
112112
) => void;
@@ -127,7 +127,7 @@ export interface WorkerListener<
127127
* world.
128128
*/
129129
progress: (
130-
job: Job<DataType, ResultType, NameType>,
130+
job: Job<Serialize<DataType>, ResultType, NameType>,
131131
progress: number | object,
132132
) => void;
133133

@@ -171,7 +171,7 @@ export class Worker<
171171

172172
private abortDelayController: AbortController | null = null;
173173
private asyncFifoQueue: AsyncFifoQueue<void | Job<
174-
DataType,
174+
Serialize<DataType>,
175175
ResultType,
176176
NameType
177177
>>;
@@ -187,7 +187,7 @@ export class Worker<
187187
private _repeat: Repeat;
188188

189189
protected paused: Promise<void>;
190-
protected processFn: Processor<DataType, ResultType, NameType>;
190+
protected processFn: Processor<Serialize<DataType>, ResultType, NameType>;
191191
protected running = false;
192192

193193
static RateLimitError(): Error {
@@ -196,7 +196,7 @@ export class Worker<
196196

197197
constructor(
198198
name: string,
199-
processor?: string | URL | null | Processor<DataType, ResultType, NameType>,
199+
processor?: string | URL | null | Processor<Serialize<DataType>, ResultType, NameType>,
200200
opts?: WorkerOptions,
201201
Connection?: typeof RedisConnection,
202202
) {
@@ -289,7 +289,7 @@ export class Worker<
289289
useWorkerThreads: this.opts.useWorkerThreads,
290290
});
291291

292-
this.processFn = sandbox<DataType, ResultType, NameType>(
292+
this.processFn = sandbox<Serialize<DataType>, ResultType, NameType>(
293293
processor,
294294
this.childPool,
295295
).bind(this);
@@ -348,7 +348,7 @@ export class Worker<
348348
}
349349

350350
protected callProcessJob(
351-
job: Job<DataType, ResultType, NameType>,
351+
job: Job<Serialize<DataType>, ResultType, NameType>,
352352
token: string,
353353
): Promise<ResultType> {
354354
return this.processFn(job, token);
@@ -357,9 +357,9 @@ export class Worker<
357357
protected createJob(
358358
data: JobJsonRaw,
359359
jobId: string,
360-
): Job<DataType, ResultType, NameType> {
360+
): Job<Serialize<DataType>, ResultType, NameType> {
361361
return this.Job.fromJSON(this as MinimalQueue, data, jobId) as Job<
362-
DataType,
362+
Serialize<DataType>,
363363
ResultType,
364364
NameType
365365
>;
@@ -423,7 +423,7 @@ export class Worker<
423423
this.startLockExtenderTimer(jobsInProgress);
424424

425425
const asyncFifoQueue = (this.asyncFifoQueue =
426-
new AsyncFifoQueue<void | Job<DataType, ResultType, NameType>>());
426+
new AsyncFifoQueue<void | Job<Serialize<DataType>, ResultType, NameType>>());
427427

428428
let tokenPostfix = 0;
429429

@@ -450,7 +450,7 @@ export class Worker<
450450
const token = `${this.id}:${tokenPostfix++}`;
451451

452452
const fetchedJob = this.retryIfFailed<void | Job<
453-
DataType,
453+
Serialize<DataType>,
454454
ResultType,
455455
NameType
456456
>>(
@@ -484,18 +484,18 @@ export class Worker<
484484

485485
// Since there can be undefined jobs in the queue (when a job fails or queue is empty)
486486
// we iterate until we find a job.
487-
let job: Job<DataType, ResultType, NameType> | void;
487+
let job: Job<Serialize<DataType>, ResultType, NameType> | void;
488488
do {
489489
job = await asyncFifoQueue.fetch();
490490
} while (!job && asyncFifoQueue.numQueued() > 0);
491491

492492
if (job) {
493493
const token = job.token;
494494
asyncFifoQueue.add(
495-
this.retryIfFailed<void | Job<DataType, ResultType, NameType>>(
495+
this.retryIfFailed<void | Job<Serialize<DataType>, ResultType, NameType>>(
496496
() =>
497497
this.processJob(
498-
<Job<DataType, ResultType, NameType>>job,
498+
<Job<Serialize<DataType>, ResultType, NameType>>job,
499499
token,
500500
() => asyncFifoQueue.numTotal() <= this.opts.concurrency,
501501
jobsInProgress,
@@ -533,7 +533,7 @@ export class Worker<
533533
bclient: RedisClient,
534534
token: string,
535535
{ block = true }: GetNextJobOptions = {},
536-
): Promise<Job<DataType, ResultType, NameType> | undefined> {
536+
): Promise<Job<Serialize<DataType>, ResultType, NameType> | undefined> {
537537
if (this.paused) {
538538
if (block) {
539539
await this.paused;
@@ -608,7 +608,7 @@ will never work with more accuracy than 1ms. */
608608
client: RedisClient,
609609
token: string,
610610
name?: string,
611-
): Promise<Job<DataType, ResultType, NameType>> {
611+
): Promise<Job<Serialize<DataType>, ResultType, NameType>> {
612612
const [jobData, id, limitUntil, delayUntil] =
613613
await this.scripts.moveToActive(client, token, name);
614614
this.updateDelays(limitUntil, delayUntil);
@@ -719,7 +719,7 @@ will never work with more accuracy than 1ms. */
719719
jobData?: JobJsonRaw,
720720
jobId?: string,
721721
token?: string,
722-
): Promise<Job<DataType, ResultType, NameType>> {
722+
): Promise<Job<Serialize<DataType>, ResultType, NameType>> {
723723
if (!jobData) {
724724
if (!this.drained) {
725725
this.emit('drained');
@@ -740,11 +740,11 @@ will never work with more accuracy than 1ms. */
740740
}
741741

742742
async processJob(
743-
job: Job<DataType, ResultType, NameType>,
743+
job: Job<Serialize<DataType>, ResultType, NameType>,
744744
token: string,
745745
fetchNextCallback = () => true,
746746
jobsInProgress: Set<{ job: Job; ts: number }>,
747-
): Promise<void | Job<DataType, ResultType, NameType>> {
747+
): Promise<void | Job<Serialize<DataType>, ResultType, NameType>> {
748748
if (!job || this.closing || this.paused) {
749749
return;
750750
}
@@ -1051,10 +1051,10 @@ will never work with more accuracy than 1ms. */
10511051

10521052
stalled.forEach((jobId: string) => this.emit('stalled', jobId, 'active'));
10531053

1054-
const jobPromises: Promise<Job<DataType, ResultType, NameType>>[] = [];
1054+
const jobPromises: Promise<Job<Serialize<DataType>, ResultType, NameType>>[] = [];
10551055
for (let i = 0; i < failed.length; i++) {
10561056
jobPromises.push(
1057-
Job.fromId<DataType, ResultType, NameType>(
1057+
Job.fromId<Serialize<DataType>, ResultType, NameType>(
10581058
this as MinimalQueue,
10591059
failed[i],
10601060
),
@@ -1069,8 +1069,8 @@ will never work with more accuracy than 1ms. */
10691069
this.notifyFailedJobs(await Promise.all(jobPromises));
10701070
}
10711071

1072-
private notifyFailedJobs(failedJobs: Job<DataType, ResultType, NameType>[]) {
1073-
failedJobs.forEach((job: Job<DataType, ResultType, NameType>) =>
1072+
private notifyFailedJobs(failedJobs: Job<Serialize<DataType>, ResultType, NameType>[]) {
1073+
failedJobs.forEach((job: Job<Serialize<DataType>, ResultType, NameType>) =>
10741074
this.emit(
10751075
'failed',
10761076
job,
@@ -1081,7 +1081,7 @@ will never work with more accuracy than 1ms. */
10811081
}
10821082

10831083
private moveLimitedBackToWait(
1084-
job: Job<DataType, ResultType, NameType>,
1084+
job: Job<Serialize<DataType>, ResultType, NameType>,
10851085
token: string,
10861086
) {
10871087
return this.scripts.moveJobFromActiveToWait(job.id, token);

‎src/types/index.ts

+1
Original file line numberDiff line numberDiff line change
@@ -5,3 +5,4 @@ export * from './job-json-sandbox';
55
export * from './job-options';
66
export * from './job-type';
77
export * from './repeat-strategy';
8+
export * from './serialize';

‎src/types/serialize.ts

+146
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
/**
2+
* The `Serialize` type emulates the operation `JSON.parse(JSON.stringify(x))` done
3+
* when passing the `data` to a `Job` in a `Worker`.
4+
*/
5+
export type Serialize<T> =
6+
/**
7+
* If the input is any then the output must as well be any
8+
*/
9+
IsAny<T> extends true
10+
? any
11+
: /**
12+
* Under the hood, JSON.stringify calls the `toJSON()` method on his parameter.
13+
*/
14+
T extends { toJSON(): infer U }
15+
? U extends JsonValue
16+
? U
17+
: unknown
18+
: /**
19+
* Primitives
20+
*/
21+
T extends JsonPrimitive
22+
? T
23+
: /**
24+
* Primitive wrappers
25+
*/
26+
T extends String
27+
? string
28+
: T extends Number
29+
? number
30+
: T extends Boolean
31+
? boolean
32+
: /**
33+
* JSON.stringify returns always `{}` for a `Promise`
34+
*/
35+
T extends Promise<unknown>
36+
? '{}'
37+
: /**
38+
* Map
39+
*/
40+
T extends Map<unknown, unknown>
41+
? EmptyObject
42+
: /**
43+
* Set
44+
*/
45+
T extends Set<unknown>
46+
? EmptyObject
47+
: /**
48+
* Array views
49+
*/
50+
T extends TypedArray
51+
? Record<string, number>
52+
: /**
53+
* Some object can't be serialized, so we remove them.
54+
*/
55+
T extends NotJson
56+
? never
57+
: /**
58+
* Arrays
59+
*/
60+
T extends []
61+
? []
62+
: T extends readonly [infer F, ...infer R]
63+
? [NeverToNull<Serialize<F>>, ...Serialize<R>]
64+
: T extends readonly unknown[]
65+
? Array<NeverToNull<Serialize<T[number]>>>
66+
: /**
67+
* Objects
68+
*/
69+
T extends Record<keyof unknown, unknown>
70+
? Prettify<SerializeObject<T>>
71+
: /**
72+
* Unknown
73+
*/
74+
unknown extends T
75+
? unknown
76+
: never;
77+
78+
/**
79+
* Some utils.
80+
*/
81+
82+
type NotJson = undefined | symbol | ((...args: any[]) => unknown);
83+
84+
// value is always not JSON => true
85+
// value is always JSON => false
86+
// value is somtimes JSON, sometimes not JSON => boolean
87+
// note: cannot be inlined as logic requires union distribution
88+
type ValueIsNotJson<T> = T extends NotJson ? true : false;
89+
90+
// note: remove optionality so that produced values are never `undefined`,
91+
// only `true`, `false`, or `boolean`
92+
type IsNotJson<T> = { [K in keyof T]-?: ValueIsNotJson<T[K]> };
93+
94+
type SerializeValues<T> = { [K in keyof T]: Serialize<T[K]> };
95+
96+
type SerializeObject<T extends Record<keyof unknown, unknown>> =
97+
// required
98+
{
99+
[K in keyof T as unknown extends T[K]
100+
? never
101+
: IsNotJson<T>[K] extends false
102+
? K
103+
: never]: SerializeValues<T>[K];
104+
} & {
105+
// optional
106+
[K in keyof T as unknown extends T[K]
107+
? K
108+
: // if the value is always JSON, then it's not optional
109+
IsNotJson<T>[K] extends false
110+
? never
111+
: // if the value is always not JSON, omit it entirely
112+
IsNotJson<T>[K] extends true
113+
? never
114+
: // if the value is mixed, then it's optional
115+
K]?: SerializeValues<T>[K];
116+
};
117+
118+
type JsonPrimitive = string | number | boolean | null;
119+
120+
type JsonArray = JsonValue[] | readonly JsonValue[];
121+
122+
type JsonObject = { [K in string]: JsonValue } & { [K in string]?: JsonValue };
123+
124+
type JsonValue = JsonPrimitive | JsonObject | JsonArray;
125+
126+
type TypedArray =
127+
| Int8Array
128+
| Uint8Array
129+
| Uint8ClampedArray
130+
| Int16Array
131+
| Uint16Array
132+
| Int32Array
133+
| Uint32Array
134+
| Float32Array
135+
| Float64Array
136+
| BigInt64Array
137+
| BigUint64Array;
138+
139+
type Prettify<T> = { [K in keyof T]: T[K] } & {};
140+
141+
type NeverToNull<T> = [T] extends [never] ? null : T;
142+
143+
declare const emptyObjectSymbol: unique symbol;
144+
type EmptyObject = { [emptyObjectSymbol]?: never };
145+
146+
type IsAny<T> = 0 extends 1 & T ? true : false;

0 commit comments

Comments
 (0)
Please sign in to comment.