Skip to content

create_subscriber

faststream.kafka.subscriber.factory.create_subscriber #

create_subscriber(
    *topics: Union[str, Topic],
    batch: bool,
    batch_timeout_ms: int,
    max_records: int | None,
    group_id: str | None,
    listener: Optional[ConsumerRebalanceListener],
    pattern: str | None,
    connection_args: dict[str, Any],
    partitions: Collection[TopicPartition],
    ack_policy: AckPolicy,
    max_workers: int,
    no_reply: bool,
    config: KafkaBrokerConfig,
    title_: str | None,
    description_: str | None,
    include_in_schema: bool,
) -> Union[
    DefaultSubscriber,
    BatchSubscriber,
    ConcurrentDefaultSubscriber,
    ConcurrentBetweenPartitionsSubscriber,
]
Source code in faststream/kafka/subscriber/factory.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def create_subscriber(
    *topics: Union[str, "Topic"],
    batch: bool,
    batch_timeout_ms: int,
    max_records: int | None,
    # Kafka information
    group_id: str | None,
    listener: Optional["ConsumerRebalanceListener"],
    pattern: str | None,
    connection_args: dict[str, Any],
    partitions: Collection["TopicPartition"],
    # Subscriber args
    ack_policy: "AckPolicy",
    max_workers: int,
    no_reply: bool,
    config: "KafkaBrokerConfig",
    # Specification args
    title_: str | None,
    description_: str | None,
    include_in_schema: bool,
) -> Union[
    "DefaultSubscriber",
    "BatchSubscriber",
    "ConcurrentDefaultSubscriber",
    "ConcurrentBetweenPartitionsSubscriber",
]:
    declared_topics = [Topic.validate(t) for t in topics]

    _validate_input_for_misconfigure(
        *declared_topics,
        pattern=pattern,
        partitions=partitions,
        ack_policy=ack_policy,
        max_workers=max_workers,
        batch=batch,
        client_rack=connection_args.get("client_rack"),
    )

    subscriber_config = KafkaSubscriberConfig(
        topics=declared_topics,
        partitions=partitions,
        connection_args=connection_args,
        group_id=group_id,
        listener=listener,
        pattern=pattern,
        no_reply=no_reply,
        _outer_config=config,
        _ack_policy=ack_policy,
    )

    calls = CallsCollection[Any]()

    specification = KafkaSubscriberSpecification(
        _outer_config=config,
        calls=calls,
        specification_config=KafkaSubscriberSpecificationConfig(
            topics=declared_topics,
            partitions=partitions,
            pattern=pattern,
            title_=title_,
            description_=description_,
            include_in_schema=include_in_schema,
        ),
    )

    if batch:
        return BatchSubscriber(
            subscriber_config,
            specification,
            calls,
            batch_timeout_ms=batch_timeout_ms,
            max_records=max_records,
        )

    if max_workers > 1:
        if subscriber_config.ack_first:
            return ConcurrentDefaultSubscriber(
                subscriber_config,
                specification,
                calls,
                max_workers=max_workers,
            )

        subscriber_config.topics = (declared_topics[0],)
        return ConcurrentBetweenPartitionsSubscriber(
            subscriber_config,
            specification,
            calls,
            max_workers=max_workers,
        )

    return DefaultSubscriber(subscriber_config, specification, calls)