# Path: cb_bin_client.py
# def decode_collection_id(key: bytes) -> Tuple[int, bytes]:
#     # A leb128 varint encodes the CID
#     data = array.array('B')
#     data.frombytes(key)
#     cid = data[0] & 0x7f
#     end = 1
#     if (data[0] & 0x80) == 0x80:
#         shift = 7
#         for end in range(1, len(data)):
#             cid |= ((data[end] & 0x7f) << shift)
#             if (data[end] & 0x80) == 0:
#                 break
#             shift = shift + 7
# 
#         end = end + 1
#         if end == len(data):
#             #  We should of stopped for a stop byte, not the end of the buffer
#             raise ValueError("encoded key did not contain a stop byte")
#     return cid, key[end:]
# 
# def encode_collection_id(cid: int) -> bytes:
#     output = array.array('B', [0])
#     while cid > 0:
#         byte = cid & 0xFF
#         cid >>= 7
#         # CID has more bits
#         if cid > 0:
#             # Set the 'continue' bit of this byte
#             byte |= 0x80
#             output[-1] = byte
#             output.append(0)
#         else:
#             output[-1] = byte
#     return output.tobytes()


import glob
import json
import logging
import os
import queue
import re
import struct
import threading
import couchstore  # pylint: disable=import-error
import couchbaseConstants
import pump
from collections import defaultdict
from typing import Any, Dict, List, Optional, Tuple
from cb_bin_client import decode_collection_id, encode_collection_id
        source_vbucket_state = \
            getattr(self.opts, 'source_vbucket_state', 'active')

        source_nodes = self.source_bucket['nodes']
        if len(source_nodes) != 1:
            self.queue.put((f'error: expected 1 node in source_bucket: {self.source_bucket["name"]}', None))
            return

        vbucket_states = source_nodes[0].get('vbucket_states', None)
        if not vbucket_states:
            self.queue.put((f'error: missing vbucket_states in source_bucket: {self.source_bucket["name"]}', None))
            return

        vbuckets = vbucket_states.get(source_vbucket_state, None)
        if vbuckets is None:  # Empty dict is valid.
            self.queue.put((f'error: missing vbuckets in source_bucket: {self.source_bucket["name"]}', None))
            return

        batch_max_size = self.opts.extra['batch_max_size']
        batch_max_bytes = self.opts.extra['batch_max_bytes']

        store = None
        vbucket_id = None

        # Level of indirection since we can't use python 3 nonlocal statement.
        abatch: List[pump.Batch] = [pump.Batch(self)]

        def change_callback(doc_info):
            if doc_info:
                # Handle the new key name spacing for collections and co

# Path: cb_bin_client.py
# def decode_collection_id(key: bytes) -> Tuple[int, bytes]:
#     # A leb128 varint encodes the CID
#     data = array.array('B')
#     data.frombytes(key)
#     cid = data[0] & 0x7f
#     end = 1
#     if (data[0] & 0x80) == 0x80:
#         shift = 7
#         for end in range(1, len(data)):
#             cid |= ((data[end] & 0x7f) << shift)
#             if (data[end] & 0x80) == 0:
#                 break
#             shift = shift + 7
# 
#         end = end + 1
#         if end == len(data):
#             #  We should of stopped for a stop byte, not the end of the buffer
#             raise ValueError("encoded key did not contain a stop byte")
#     return cid, key[end:]
# 
# def encode_collection_id(cid: int) -> bytes:
#     output = array.array('B', [0])
#     while cid > 0:
#         byte = cid & 0xFF
#         cid >>= 7
#         # CID has more bits
#         if cid > 0:
#             # Set the 'continue' bit of this byte
#             byte |= 0x80
#             output[-1] = byte
#             output.append(0)
#         else:
#             output[-1] = byte
#     return output.tobytes()


# Path: pump_sfd.py
import glob
import json
import logging
import os
import queue
import re
import struct
import threading
import couchstore  # pylint: disable=import-error
import couchbaseConstants
import pump
from collections import defaultdict
from typing import Any, Dict, List, Optional, Tuple
from cb_bin_client import decode_collection_id, encode_collection_id

        source_vbucket_state = \
            getattr(self.opts, 'source_vbucket_state', 'active')

        source_nodes = self.source_bucket['nodes']
        if len(source_nodes) != 1:
            self.queue.put((f'error: expected 1 node in source_bucket: {self.source_bucket["name"]}', None))
            return

        vbucket_states = source_nodes[0].get('vbucket_states', None)
        if not vbucket_states:
            self.queue.put((f'error: missing vbucket_states in source_bucket: {self.source_bucket["name"]}', None))
            return

        vbuckets = vbucket_states.get(source_vbucket_state, None)
        if vbuckets is None:  # Empty dict is valid.
            self.queue.put((f'error: missing vbuckets in source_bucket: {self.source_bucket["name"]}', None))
            return

        batch_max_size = self.opts.extra['batch_max_size']
        batch_max_bytes = self.opts.extra['batch_max_bytes']

        store = None
        vbucket_id = None

        # Level of indirection since we can't use python 3 nonlocal statement.
        abatch: List[pump.Batch] = [pump.Batch(self)]

        def change_callback(doc_info):
            if doc_info:
                # Handle the new key name spacing for collections and co

                cid, key = decode_collection_id(doc_info.id.encode())