subscribe_to_shard ¶

def subscribe_to_shard(
    self,
    consumer_arn: "capo_kinesis.types.consumer_arn.ConsumerARN",
    shard_id: "capo_kinesis.types.shard_id.ShardId",
    starting_position: "capo_kinesis.types.starting_position.StartingPosition",
    *,
    config_overrides: Optional[KinesisClientConfig] = None,
    stream_id: Optional["capo_kinesis.types.stream_id.StreamId"] = None,
    dry_run: Optional["capo_kinesis.types.boolean_object.BooleanObject"] = None,
) -> "capo_kinesis.types.subscribe_to_shard_output.SubscribeToShardOutput":
    """<p>This operation establishes an HTTP/2 connection between the consumer you specify in the <code>ConsumerARN</code> parameter and the shard you specify in the <code>ShardId</code> parameter. After the connection is successfully established, Kinesis Data Streams pushes records from the shard to the consumer over this connection. Before you call this operation, call <a>RegisterStreamConsumer</a> to register the consumer with Kinesis Data Streams.</p> <p>When the <code>SubscribeToShard</code> call succeeds, your consumer starts receiving events of type <a>SubscribeToShardEvent</a> over the HTTP/2 connection for up to 5 minutes, after which time you need to call <code>SubscribeToShard</code> again to renew the subscription if you want to continue to receive records.</p> <p>You can make one call to <code>SubscribeToShard</code> per second per registered consumer per shard. For example, if you have a 4000 shard stream and two registered stream consumers, you can make one <code>SubscribeToShard</code> request per second for each combination of shard and registered consumer, allowing you to subscribe both consumers to all 4000 shards in one second. </p> <p>If you call <code>SubscribeToShard</code> again with the same <code>ConsumerARN</code> and <code>ShardId</code> within 5 seconds of a successful call, you'll get a <code>ResourceInUseException</code>. If you call <code>SubscribeToShard</code> 5 seconds or more after a successful call, the second call takes over the subscription and the previous connection expires or fails with a <code>ResourceInUseException</code>.</p> <p>For an example of how to use this operation, see <a href="https://docs.aws.amazon.com/streams/latest/dev/building-enhanced-consumers-api.html">Enhanced Fan-Out Using the Kinesis Data Streams API</a>.</p>

    Args:
        consumer_arn: <p>For this parameter, use the value you obtained when you called <a>RegisterStreamConsumer</a>.</p>
        stream_id: <p>Not Implemented. Reserved for future use.</p>
        shard_id: <p>The ID of the shard you want to subscribe to. To see a list of all the shards for a given stream, use <a>ListShards</a>.</p>
        starting_position: <p>The starting position in the data stream from which to start streaming.</p>
        dry_run: <p>Checks if your request will succeed. <code>DryRun</code> is an optional parameter.</p>

    Raises:
        capo_kinesis.errors.access_denied_exception.AccessDeniedException: <p>Specifies that you do not have the permissions required to perform this operation.</p>
        capo_kinesis.errors.dry_run_operation_exception.DryRunOperationException: <p>The request was rejected because the DryRun parameter was specified.</p>
        capo_kinesis.errors.invalid_argument_exception.InvalidArgumentException: <p>A specified parameter exceeds its restrictions, is not supported, or can't be used. For more information, see the returned message.</p>
        capo_kinesis.errors.limit_exceeded_exception.LimitExceededException: <p>The requested resource exceeds the maximum number allowed, or the number of concurrent stream requests exceeds the maximum number allowed. </p>
        capo_kinesis.errors.resource_in_use_exception.ResourceInUseException: <p>The resource is not available for this operation. For successful operation, the resource must be in the <code>ACTIVE</code> state.</p>
        capo_kinesis.errors.resource_not_found_exception.ResourceNotFoundException: <p>The requested resource could not be found. The stream might not be specified correctly.</p>
        capo_kinesis.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_kinesis.types.subscribe_to_shard_input.SubscribeToShardInput]",
    ) -> OperationResponse[
        "capo_kinesis.types.subscribe_to_shard_output.SubscribeToShardOutput"
    ]:
        import capo_kinesis._operations.kinesis_20131202.subscribe_to_shard

        output, http_response = (
            capo_kinesis._operations.kinesis_20131202.subscribe_to_shard.subscribe_to_shard(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_output(
        output: "capo_kinesis.types.subscribe_to_shard_output.SubscribeToShardOutput",
    ) -> dict[str, Any]:
        import capo_kinesis.types.subscribe_to_shard_output

        return capo_kinesis.types.subscribe_to_shard_output.redact_sensitive(output)

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("SubscribeToShard", redact_output=_redact_output),
    )
    input_: capo_kinesis.types.subscribe_to_shard_input.SubscribeToShardInput = {
        "consumer_arn": consumer_arn,
        "shard_id": shard_id,
        "starting_position": starting_position,
    }
    if stream_id is not None:
        input_["stream_id"] = stream_id
    if dry_run is not None:
        input_["dry_run"] = dry_run

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    return response.output