Rpc (bonsai_web.bonsai_web.rpc_effect_kernel.Rpc_effect

val dispatcher : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  ('query, 'response) Async_rpc_kernel.Rpc.Rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  Bonsai.graph @ local ->
  ('query -> 'response Core.Or_error.t Bonsai.For_open.Effect.t) Bonsai.t

An effect for sending a particular RPC to a particular place.

where_to_connect defaults to self ~on_conn_failure:Retry_until_success.

val babel_dispatcher : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  ('query -> 'response Core.Or_error.t Async_kernel.Deferred.t) Babel.Caller.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  Bonsai.graph @ local ->
  ('query -> 'response Core.Or_error.t Bonsai.For_open.Effect.t) Bonsai.t

val streamable_dispatcher : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  ('query, 'response) Streamable.Plain_rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  Bonsai.graph @ local ->
  ('query -> 'response Core.Or_error.t Bonsai.For_open.Effect.t) Bonsai.t

val poll : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query, 'response) Async_rpc_kernel.Rpc.Rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  every:Core.Time_ns.Span.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

A computation that periodically dispatches on an RPC and keeps track of the most recent response. Only one request will be in-flight at any point in time.

clear_when_deactivated determines whether the most recent response should be discarded when the component is deactivated. Default is true.

where_to_connect defaults to self ~on_conn_failure:Surface_error_to_rpc.

val babel_poll : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query -> 'response Core.Or_error.t Async_kernel.Deferred.t) Babel.Caller.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  every:Core.Time_ns.Span.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

Analagous to poll for babel RPCs. See poll for details.

val streamable_poll : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query, 'response) Streamable.Plain_rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  every:Core.Time_ns.Span.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

Analagous to poll for Streamable plain RPCs. See poll for details.

val streamable_poll_until_ok : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query, 'response) Streamable.Plain_rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  retry_interval:Core.Time_ns.Span.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

Analagous to poll_until_ok for Streamable plain RPCs. See poll_until_ok for details.

val shared_poller : 
  here:lexing_position ->
  ('query, _) Core.Comparator.Module.t ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query, 'response) Async_rpc_kernel.Rpc.Rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  every:Core.Time_ns.Span.t Bonsai.t ->
  Bonsai.graph @ local ->
  ('query, 'response) Shared_poller.t Bonsai.t

val shared_babel_poller : 
  here:lexing_position ->
  ('query, _) Core.Comparator.Module.t ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query -> 'response Core.Or_error.t -> unit Bonsai.Effect.t) Bonsai.t ->
  ('query -> 'response Core.Or_error.t Async_kernel.Deferred.t) Babel.Caller.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  every:Core.Time_ns.Span.t Bonsai.t ->
  Bonsai.graph @ local ->
  ('query, 'response) Shared_poller.t Bonsai.t

val poll_until_ok : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query, 'response) Async_rpc_kernel.Rpc.Rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  retry_interval:Core.Time_ns.Span.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

Like poll, but stops polling the same input query after an ok response. If the query changes, the computation will resume polling until it receives another ok response. If the computation receives an error response, it will retry sending the RPC after waiting retry_interval.

val poll_until_condition_met : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query, 'response) Async_rpc_kernel.Rpc.Rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  every:Core.Time_ns.Span.t Bonsai.t ->
  condition:('response -> [ `Continue | `Stop_polling ]) Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

Similar to poll_until_ok, but will continue polling until condition response indicates to `Stop_polling on an ok response. Also like poll_until_ok, it will resume polling when the query changes, the condition changes, or the computation receives an error response.

val babel_poll_until_ok : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query -> 'response Core.Or_error.t Async_kernel.Deferred.t) Babel.Caller.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  retry_interval:Core.Time_ns.Span.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

val babel_poll_until_condition_met : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query -> 'response Core.Or_error.t Async_kernel.Deferred.t) Babel.Caller.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  every:Core.Time_ns.Span.t Bonsai.t ->
  condition:('response -> [ `Continue | `Stop_polling ]) Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  'query Bonsai.t ->
  Bonsai.graph @ local ->
  'output Bonsai.t

val manual_poll : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query, 'response) Async_rpc_kernel.Rpc.Rpc.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  Bonsai.graph @ local ->
  ('output * ('query -> 'response Core.Or_error.t Bonsai.For_open.Effect.t))
    Bonsai.t

Like poll, but returns the poll result and querying effect separately. This allows for more flexible composition and custom handling of the state. The returned effect can be scheduled to send/re-send the RPC.

Unlike poll, this function does not automatically schedule the effect - no polling happens unless you explicitly schedule the returned effect yourself.

val babel_manual_poll : 
  here:lexing_position ->
  ?sexp_of_query:('query -> Core.Sexp.t) ->
  ?sexp_of_response:('response -> Core.Sexp.t) ->
  equal_query:('query -> 'query -> bool) ->
  ?equal_response:('response -> 'response -> bool) ->
  ?clear_when_deactivated:bool ->
  ?intercept_query:
    ('query -> Inflight_query_id.t -> 'query Bonsai.For_open.Effect.t) Bonsai.t ->
  ?on_response_received:
    ('query ->
      'response Core.Or_error.t ->
      unit Bonsai.For_open.Effect.t)
      Bonsai.t ->
  ('query -> 'response Core.Or_error.t Async_kernel.Deferred.t) Babel.Caller.t ->
  where_to_connect:Where_to_connect.t Bonsai.t ->
  output_type:('query, 'response, 'output) Poll_result.Output_type.t ->
  Bonsai.graph @ local ->
  ('output * ('query -> 'response Core.Or_error.t Bonsai.For_open.Effect.t))
    Bonsai.t

Analagous to manual_poll for babel RPCs. See manual_poll for details.