def stream_slices(self, stream_state: Mapping[(str, Any)]=None, **kwargs) -> Iterable[Optional[Mapping[(str, any)]]]:
    "\n        The logic for incrementally syncing threads is not very obvious, so buckle up.\n\n        To get all messages in a thread, one must specify the channel and timestamp of the parent (first) message of that thread, basically its ID.\n\n        One complication is that threads can be updated at any time in the future. Therefore, if we wanted to comprehensively sync data i.e: get every\n        single response in a thread, we'd have to read every message in the slack instance every time we ran a sync, because otherwise there is no\n        way to guarantee that a thread deep in the past didn't receive a new message.\n\n        A pragmatic workaround is to say we want threads to be at least N days fresh i.e: look back N days into the past, get every message since,\n        and read all of the thread responses. This is essentially the approach we're taking here via slicing: create slices from N days into the\n        past and read all messages in threads since then. We could optionally filter out records we have already read, but that's omitted to keep\n        the logic simple to reason about.\n\n        Good luck.\n        "
    stream_state = (stream_state or {})
    channels_stream = Channels(authenticator=self.authenticator)
    if (self.cursor_field in stream_state):
        messages_start_date = (pendulum.from_timestamp(stream_state[self.cursor_field]) - self.messages_lookback_window)
    else:
        messages_start_date = pendulum.from_timestamp(self._start_ts)
    messages_stream = ChannelMessages(authenticator=self.authenticator, default_start_date=messages_start_date)
    for message_chunk in messages_stream.stream_slices(stream_state={self.cursor_field: messages_start_date.timestamp()}):
        self.logger.info(f'Syncing replies {message_chunk}')
        for channel in channels_stream.read_records(sync_mode=SyncMode.full_refresh):
            message_chunk['channel'] = channel['id']
            for message in messages_stream.read_records(sync_mode=SyncMode.full_refresh, stream_slice=message_chunk):
                (yield {'channel': channel['id'], self.cursor_field: message[self.primary_key]})
The logic for incrementally syncing threads is not very obvious, so buckle up.

To get all messages in a thread, one must specify the channel and timestamp of the parent (first) message of that thread, basically its ID.

One complication is that threads can be updated at any time in the future. Therefore, if we wanted to comprehensively sync data i.e: get every
single response in a thread, we'd have to read every message in the slack instance every time we ran a sync, because otherwise there is no
way to guarantee that a thread deep in the past didn't receive a new message.

A pragmatic workaround is to say we want threads to be at least N days fresh i.e: look back N days into the past, get every message since,
and read all of the thread responses. This is essentially the approach we're taking here via slicing: create slices from N days into the
past and read all messages in threads since then. We could optionally filter out records we have already read, but that's omitted to keep
the logic simple to reason about.

Good luck.
def stream_slices(self, stream_state: Mapping[(str, Any)]=None, **kwargs) -> Iterable[Optional[Mapping[(str, any)]]]:
    "\n        The logic for incrementally syncing threads is not very obvious, so buckle up.\n\n        To get all messages in a thread, one must specify the channel and timestamp of the parent (first) message of that thread, basically its ID.\n\n        One complication is that threads can be updated at any time in the future. Therefore, if we wanted to comprehensively sync data i.e: get every\n        single response in a thread, we'd have to read every message in the slack instance every time we ran a sync, because otherwise there is no\n        way to guarantee that a thread deep in the past didn't receive a new message.\n\n        A pragmatic workaround is to say we want threads to be at least N days fresh i.e: look back N days into the past, get every message since,\n        and read all of the thread responses. This is essentially the approach we're taking here via slicing: create slices from N days into the\n        past and read all messages in threads since then. We could optionally filter out records we have already read, but that's omitted to keep\n        the logic simple to reason about.\n\n        Good luck.\n        "
    stream_state = (stream_state or {})
    channels_stream = Channels(authenticator=self.authenticator)
    if (self.cursor_field in stream_state):
        messages_start_date = (pendulum.from_timestamp(stream_state[self.cursor_field]) - self.messages_lookback_window)
    else:
        messages_start_date = pendulum.from_timestamp(self._start_ts)
    messages_stream = ChannelMessages(authenticator=self.authenticator, default_start_date=messages_start_date)
    for message_chunk in messages_stream.stream_slices(stream_state={self.cursor_field: messages_start_date.timestamp()}):
        self.logger.info(f'Syncing replies {message_chunk}')
        for channel in channels_stream.read_records(sync_mode=SyncMode.full_refresh):
            message_chunk['channel'] = channel['id']
            for message in messages_stream.read_records(sync_mode=SyncMode.full_refresh, stream_slice=message_chunk):
                (yield {'channel': channel['id'], self.cursor_field: message[self.primary_key]})