class_docstr: HTTPLister is defined to work with the HTTPEventProvider. It implements two
  APIs, poll_for_event() and event_checkpointed(). An instance of HTTPListener will
  be started by the get_message() of the workflow.wait_for_event() to listen for
  an event from the HTTPEventProvider instance (a Ray Serve deployment). Another
  instance of HTTPListener will be started by the message_committed() of the
  workflow.wait_for_event() to confirm that the event has been checkpointed.
  
  
  Example definition
  ==================
  
  ```
  class HTTPListener:
  
      def __init__(self):
  
      async def poll_for_event(self, event_key) -> Event:
  
      async def event_checkpointed(self, event) -> None:
  
  ```
  
  Example Usage
  =============
  
  .. testcode::
  
      import tempfile
      from ray import workflow
      from ray.workflow.http_event_provider import HTTPListener
  
      temp_dir = tempfile.TemporaryDirectory()
      ray.init(storage=f"file://{temp_dir.name}")
  
      serve.start(detached=True)
      event_node = workflow.wait_for_event(HTTPListener, event_key='')
  
      @ray.remote
      def handle_event(arg):
          return arg
  
      workflow.run_async(handle_event.bind(event_node), workflow_id="http_listener")
class_docstr_tok: ["HTTPLister","is","defined","to","work","with","the","HTTPEventProvider",".","It","implements","two","APIs",",","poll_for_event","(",")","and","event_checkpointed","(",")",".","An","instance","of","HTTPListener","will","be","started","by","the","get_message","(",")","of","the","workflow.wait_for_event","(",")","to","listen","for","an","event","from","the","HTTPEventProvider","instance","(","a","Ray","Serve","deployment",")",".","Another","instance","of","HTTPListener","will","be","started","by","the","message_committed","(",")","of","the","workflow.wait_for_event","(",")","to","confirm","that","the","event","has","been","checkpointed",".","Example","definition","==================","``","`","class","HTTPListener",":","def","__init__","(","self",")",":","async","def","poll_for_event","(","self",",","event_key",")","-",">","Event",":","async","def","event_checkpointed","(","self",",","event",")","-",">","None",":","``","`","Example","Usage","=============","..","testcode",":",":","import","tempfile","from","ray","import","workflow","from","ray.workflow.http_event_provider","import","HTTPListener","temp_dir","=","tempfile.TemporaryDirectory","(",")","ray.init","(","storage=f","''","file",":","\/\/","{","temp_dir.name","}","''",")","serve.start","(","detached=True",")","event_node","=","workflow.wait_for_event","(","HTTPListener",",","event_key=","''",")","@","ray.remote","def","handle_event","(","arg",")",":","return","arg","workflow.run_async","(","handle_event.bind","(","event_node",")",",","workflow_id=","''","http_listener","''",")"]
code: def __init__(self):
      super().__init__()
      try:
          self.handle = ray.serve.get_app_handle(
              common.HTTP_EVENT_PROVIDER_NAME
          )
      except ray.serve.exceptions.RayServeException:
          mgr = workflow_access.get_management_actor()
          ray.get(mgr.create_http_event_provider.remote())
          self.handle = ray.serve.get_app_handle(
              common.HTTP_EVENT_PROVIDER_NAME
          )
  
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 HTTPListener object from inherited base classes: EventListener with example: obj = HTTPListener()