Add support for cloudEvents

chris-aeviator commented 1 year ago

What would you like to see added? Cloudevents (https://cloudevents.io/) is a specification for describing event data in a common way. Cloud events are used in a variety of application platforms and are independent of the underlying transport layer, so plumber would be the ideal tool to send them from a terminal.

Use Case

When developing a service that expects cloudevents structured messages, plumber would act as the swiss army knife for sending them in an easy way.

Supported adapters are e.g.

kafka
mqtt
NATS
AMQP

more at https://github.com/cloudevents/spec#cloudevents-documents

Proposed Change

add a CLI flag --cloudEvent that turns --input {"data": {"userSignedUpEvent": ...} into a cloudEvent

Who Benefits from the Change(s)?

All users working with the serverlessworkflow.io specification and cloudEvent users in general.

Additional context example of a cloud event message

{
  "data":{ EVENT_DATA

 }
  "datacontenttype": "application/json; charset=utf-8",
  "id": "MESSAGE_ID

",
  "source": "//cloudaudit.googleapis.com/projects/PROJECT_ID

/logs/data_access",
  "specversion": "1.0",
  "type": "google.cloud.audit.log.v1.written",
  "time": "EVENT_GENERATION_TIME

",
  "dataschema": "https://googleapis.github.io/google-cloudevents/jsonschema/google/events/cloud/audit/v1/LogEntryData.json",
  "methodName": "jobservice.jobcompleted",
  "resourceName": "projects/my-project/jobs/bqjob_r3ac45813612fa2d6_0000017d591922c9_1",
  "serviceName": "bigquery.googleapis.com",
  "subject": "bigquery.googleapis.com/projects/my-project/jobs/bqjob_r3ac45813612fa2d6_0000017d591922c9_1",
}

sample python code to construct a cloud event

import datetime
import json
import uuid

event = {}
event['id'] = str(uuid.uuid4())
event['source'] = 'my-event-source'
event[type] = 'my-event-type'
event['specversion'] = '0.3'
event['time'] = datetime.datetime.utcnow().isoformat('T') + 'Z'
event['data'] = 'Hello World!'
event_json_str = json.dumps(event)

# Alternatively with Cloud Events SDK
from cloudevents.sdk.event import v03

event = (
    v03.Event().
    SetEventID('my-event-id').
    SetSource('my-event-source').
    SetEventType('my-event-type').
    SetEventTime(datetime.datetime.utcnow().isoformat('T') + 'Z').
    SetData('Hello World!')