Added interface type for Cloudflare pipelines binding #2654

hhoughgg · 2024-09-04T15:36:42Z

No description provided.

a-robinson · 2024-09-04T16:54:02Z

types/defines/pipeline-transform.d.ts

@@ -11,3 +11,18 @@ export abstract class PipelineTransform {
   */
  public transformJson(data: object[]): Promise<object[]>;
 }
+
+export type PipelineResponse = {


Sorry for commenting on API design here given that the code is probably already written, but what is this return type communicating that wouldn't be equally-well communicated by a rejected promise / thrown Error?

In what cases will success be false? In what cases will error be filled in vs the Promise being rejected?

I'd have expected send to just return a Promise<void> unless there was something more than success/error to be returned.

For example that's what the Queues send() method returns: https://github.com/cloudflare/workerd/blob/main/src/workerd/api/queue.h#L68

types/defines/pipeline-transform.d.ts

elithrar · 2024-09-04T16:58:09Z

types/defines/pipeline-transform.d.ts

+   * @param data The data to be send
+   * @returns A promise containing the outcome of the send
+   */
+  send(data: object[]): Promise<PipelineResponse>


Why an object - not every object is serializable? Does this cause issues with writing objects out to storage / consuming from a non-JS system?

How does this fit with Queues?

workerd/src/workerd/api/queue.h

Lines 51 to 87 in c9eca14

struct MessageSendRequest {

jsg::JsRef<jsg::JsValue> body;

// contentType determines the serialization format of the message.

jsg::Optional<kj::String> contentType;

// The number of seconds to delay the delivery of the message being sent.

jsg::Optional<int> delaySeconds;

JSG_STRUCT(body, contentType, delaySeconds);

JSG_STRUCT_TS_OVERRIDE(MessageSendRequest<Body = unknown> {

body: Body;

contentType?: QueueContentType;

});

// NOTE: Any new fields added to SendOptions must also be added here.

};

kj::Promise<void> send(jsg::Lock& js, jsg::JsValue body, jsg::Optional<SendOptions> options);

kj::Promise<void> sendBatch(jsg::Lock& js,

jsg::Sequence<MessageSendRequest> batch,

jsg::Optional<SendBatchOptions> options);

JSG_RESOURCE_TYPE(WorkerQueue) {

JSG_METHOD(send);

JSG_METHOD(sendBatch);

JSG_TS_ROOT();

JSG_TS_OVERRIDE(Queue<Body = unknown> {

send(message: Body, options?: QueueSendOptions): Promise<void>;

sendBatch(messages

: Iterable<MessageSendRequest<Body>>, options ?: QueueSendBatchOptions)

: Promise<void>;

});

JSG_TS_DEFINE(type QueueContentType = "text" | "bytes" | "json" | "v8");

}

This will act in the same way as a queue send in that the object must be structured cloneable since this is an RPC call. The only difference is the limit is 1mb instead of 128kb. https://developers.cloudflare.com/queues/configuration/javascript-apis/#queue

I think the idea behind using an object is that we want to make sure all the data in Pipelines is a row style layout. Rows can also be represented in other languages no problem and conversions of rows/columns to other formats is a well worn path.

github-actions · 2024-09-04T22:49:02Z

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

oliy · 2024-09-05T16:51:41Z

I have read the CLA Document and I hereby sign the CLA

I have read the CLA Document and I hereby sign the CLA

oliy · 2024-09-05T16:55:13Z

recheck

fhanau · 2024-09-05T17:41:12Z

@oliy Not sure why the CLA approval didn't work – maybe try commenting without the leading quote?

oliy · 2024-09-05T19:25:44Z

I have read the CLA Document and I hereby sign the CLA

oliy · 2024-09-05T19:26:02Z

recheck

hhoughgg requested review from a team as code owners September 4, 2024 15:36

hhoughgg requested review from anonrig and fhanau September 4, 2024 15:36

a-robinson reviewed Sep 4, 2024

View reviewed changes

types/defines/pipeline-transform.d.ts Outdated Show resolved Hide resolved

a-robinson reviewed Sep 4, 2024

View reviewed changes

types/defines/pipeline-transform.d.ts Outdated Show resolved Hide resolved

elithrar reviewed Sep 4, 2024

View reviewed changes

Added interface type for cloudflare pipelines binding

1ed0dea

hhoughgg force-pushed the hhoughgg/pipelines-binding branch from d3198dd to 7d8c7e3 Compare September 4, 2024 23:07

Adding pipelines binding types

02cc862

hhoughgg force-pushed the hhoughgg/pipelines-binding branch from 7d8c7e3 to 02cc862 Compare September 4, 2024 23:08

jasnell approved these changes Sep 5, 2024

View reviewed changes

a-robinson approved these changes Sep 5, 2024

View reviewed changes

github-actions bot added a commit that referenced this pull request Sep 5, 2024

@oliy has signed the CLA in #2654

f373746

a-robinson merged commit 4faf04b into cloudflare:main Sep 5, 2024
9 of 10 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added interface type for Cloudflare pipelines binding #2654

Added interface type for Cloudflare pipelines binding #2654

hhoughgg commented Sep 4, 2024

a-robinson Sep 4, 2024

a-robinson Sep 4, 2024

elithrar Sep 4, 2024

hhoughgg Sep 4, 2024

github-actions bot commented Sep 4, 2024 •

edited

Loading

oliy commented Sep 5, 2024

oliy commented Sep 5, 2024

fhanau commented Sep 5, 2024

oliy commented Sep 5, 2024

oliy commented Sep 5, 2024

	struct MessageSendRequest {
	jsg::JsRef<jsg::JsValue> body;

	// contentType determines the serialization format of the message.
	jsg::Optional<kj::String> contentType;

	// The number of seconds to delay the delivery of the message being sent.
	jsg::Optional<int> delaySeconds;

	JSG_STRUCT(body, contentType, delaySeconds);
	JSG_STRUCT_TS_OVERRIDE(MessageSendRequest<Body = unknown> {
	body: Body;
	contentType?: QueueContentType;
	});
	// NOTE: Any new fields added to SendOptions must also be added here.
	};

	kj::Promise<void> send(jsg::Lock& js, jsg::JsValue body, jsg::Optional<SendOptions> options);

	kj::Promise<void> sendBatch(jsg::Lock& js,
	jsg::Sequence<MessageSendRequest> batch,
	jsg::Optional<SendBatchOptions> options);

	JSG_RESOURCE_TYPE(WorkerQueue) {
	JSG_METHOD(send);
	JSG_METHOD(sendBatch);

	JSG_TS_ROOT();
	JSG_TS_OVERRIDE(Queue<Body = unknown> {
	send(message: Body, options?: QueueSendOptions): Promise<void>;
	sendBatch(messages
	: Iterable<MessageSendRequest<Body>>, options ?: QueueSendBatchOptions)
	: Promise<void>;
	});
	JSG_TS_DEFINE(type QueueContentType = "text" \| "bytes" \| "json" \| "v8");
	}

Added interface type for Cloudflare pipelines binding #2654

Added interface type for Cloudflare pipelines binding #2654

Conversation

hhoughgg commented Sep 4, 2024

a-robinson Sep 4, 2024

Choose a reason for hiding this comment

a-robinson Sep 4, 2024

Choose a reason for hiding this comment

elithrar Sep 4, 2024

Choose a reason for hiding this comment

hhoughgg Sep 4, 2024

Choose a reason for hiding this comment

github-actions bot commented Sep 4, 2024 • edited Loading

oliy commented Sep 5, 2024

oliy commented Sep 5, 2024

fhanau commented Sep 5, 2024

oliy commented Sep 5, 2024

oliy commented Sep 5, 2024

github-actions bot commented Sep 4, 2024 •

edited

Loading