class_docstr: HTTPEventProvider is defined to be a Ray Serve deployment with route_prefix='/event',
  which will receive external events via an HTTP endpoint. It supports FastAPI,
  e.g. post. It responds to both poll_for_event() and event_checkpointed() from
  an HTTPListener instance.
  
  HTTPListener class is designed to work with the current workflow.wait_for_event()
  implementation, where an HTTPListener instance will be initiated by the
  get_message() and message_committed() of the workflow.wait_for_event().
  
  HTTPEventProvider requires an event to arrive after HTTPListner registers
  its event_key. If an event arrived before the registration, it returns HTTP
  error code 404 with the error "workflow_id and event_key need to be registered
  to receive event. Please make sure they are registered before resending."
  
  Example definition
  ==================
  
  ```
  class HTTPEventProvider:
  
      def __init__(self):
  
      @app.post("/send_event/{workflow_id}")
      async def send_event(self, workflow_id: str, req: Request):
          Receive an external event message and acknowledge if it was processed
          by the workflow
      async def get_event_payload(self, workflow_id, event_key):
          Internal method used by HTTPListner to subscribe to an event matched by
          workflow_id and event_key
      async def report_checkpointed(self, workflow_id, event, confirmation):
          Internal method used by HTTPListner to confirm the received event has been
          checkpointed by workflow
  ```
  
  Example Usage
  =============
  .. testcode::
      :skipif: True
  
      from ray.workflow.http_event_provider import HTTPEventProvider, HTTPListener
      ray.init(address='auto', namespace='serve')
      serve.start(detached=True)
      event_node = workflow.wait_for_event(
          HTTPListener, event_key='')
      handle_event = ...
      workflow.run_aync(handle_event.bind(event_node))
  
  On a separate python process, it sends an event to the HTTPEventProvider:
  
  .. testcode::
      :skipif: True
  
      import requests
      resp = requests.post('http://127.0.0.1:8000/event/send_event/{workflow_id}',
          json={'event_key':'my_key','event_payload':'testMessage'})
class_docstr_tok: ["HTTPEventProvider","is","defined","to","be","a","Ray","Serve","deployment","with","route_prefix='\/event","'",",","which","will","receive","external","events","via","an","HTTP","endpoint",".","It","supports","FastAPI",",","e.g",".","post",".","It","responds","to","both","poll_for_event","(",")","and","event_checkpointed","(",")","from","an","HTTPListener","instance",".","HTTPListener","class","is","designed","to","work","with","the","current","workflow.wait_for_event","(",")","implementation",",","where","an","HTTPListener","instance","will","be","initiated","by","the","get_message","(",")","and","message_committed","(",")","of","the","workflow.wait_for_event","(",")",".","HTTPEventProvider","requires","an","event","to","arrive","after","HTTPListner","registers","its","event_key",".","If","an","event","arrived","before","the","registration",",","it","returns","HTTP","error","code","404","with","the","error","``","workflow_id","and","event_key","need","to","be","registered","to","receive","event",".","Please","make","sure","they","are","registered","before","resending",".","''","Example","definition","==================","``","`","class","HTTPEventProvider",":","def","__init__","(","self",")",":","@","app.post","(","``","\/send_event\/","{","workflow_id","}","''",")","async","def","send_event","(","self",",","workflow_id",":","str",",","req",":","Request",")",":","Receive","an","external","event","message","and","acknowledge","if","it","was","processed","by","the","workflow","async","def","get_event_payload","(","self",",","workflow_id",",","event_key",")",":","Internal","method","used","by","HTTPListner","to","subscribe","to","an","event","matched","by","workflow_id","and","event_key","async","def","report_checkpointed","(","self",",","workflow_id",",","event",",","confirmation",")",":","Internal","method","used","by","HTTPListner","to","confirm","the","received","event","has","been","checkpointed","by","workflow","``","`","Example","Usage","=============","..","testcode",":",":",":","skipif",":","True","from","ray.workflow.http_event_provider","import","HTTPEventProvider",",","HTTPListener","ray.init","(","address='auto","'",",","namespace='serve","'",")","serve.start","(","detached=True",")","event_node","=","workflow.wait_for_event","(","HTTPListener",",","event_key=","''",")","handle_event","=","...","workflow.run_aync","(","handle_event.bind","(","event_node",")",")","On","a","separate","python","process",",","it","sends","an","event","to","the","HTTPEventProvider",":","..","testcode",":",":",":","skipif",":","True","import","requests","resp","=","requests.post","(","'http",":","\/\/127.0.0.1:8000\/event\/send_event\/","{","workflow_id","}","'",",","json=","{","'event_key","'",":","'my_key","'",",","'event_payload","'",":","'testMessage","'","}",")"]
docstr: Maintain two data structures to track pending events and confirmations
  event_key_payload: for each registered workflow_id and event_key,
      keep the Future to be set after an event is received.
  event_checkpoint_pending: for each received event_key, keep its Future
      after checkpointing is confirmed so HTTP 200 can be returned.
docstr_tok: ["Maintain","two","data","structures","to","track","pending","events","and","confirmations","event_key_payload",":","for","each","registered","workflow_id","and","event_key",",","keep","the","Future","to","be","set","after","an","event","is","received",".","event_checkpoint_pending",":","for","each","received","event_key",",","keep","its","Future","after","checkpointing","is","confirmed","so","HTTP","200","can","be","returned","."]
code: def __init__(self):
      """Maintain two data structures to track pending events and confirmations
      event_key_payload: for each registered workflow_id and event_key,
          keep the Future to be set after an event is received.
      event_checkpoint_pending: for each received event_key, keep its Future
          after checkpointing is confirmed so HTTP 200 can be returned.
      """
      self.event_key_payload: Dict[str, Dict[str, asyncio.Future]] = {}
      self.event_checkpoint_pending: Dict[str, asyncio.Future] = {}
  
file_path: ray/python/ray/workflow/http_event_provider.py
imports: import asyncio
  from typing import Dict
  from fastapi import FastAPI, Request
  from fastapi.responses import JSONResponse
  import ray
  from ray import serve
  from ray.workflow import common, workflow_context, workflow_access
  from ray.workflow.event_listener import EventListener
  from ray.workflow.common import Event
  import logging
label_desc: Use image node_id 1 to create a new HTTPEventProvider object  with example: obj = HTTPEventProvider()