xinjinheng · xinjinheng · Nov 30, 2025
diff --git a/debug_import.py b/debug_import.py
@@ -0,0 +1,10 @@
+#!/usr/bin/env python
+
+import traceback
+
+try:
+    import msgpack.chunked
+    print('Import successful!')
+except Exception as e:
+    print(f'Import failed: {e}')
+    traceback.print_exc()
diff --git a/msgpack/__init__.py b/msgpack/__init__.py
@@ -3,6 +3,7 @@
 
 from .exceptions import *  # noqa: F403
 from .ext import ExtType, Timestamp
+from . import chunked
 
 version = (1, 1, 2)
 __version__ = "1.1.2"

diff --git a/msgpack/chunked.py b/msgpack/chunked.py
@@ -0,0 +1,248 @@
+"""Chunked MessagePack streaming with verification and resume support"""
+
+import struct
+import hashlib
+from .fallback import Unpacker, Packer
+from .exceptions import ChunkVerifyError, BreakpointCorruptError, FormatError
+
+
+class ChunkedUnpacker(Unpacker):
+    """Streaming unpacker with chunk verification and resume support"""
+
+    def __init__(
+        self,
+        file_like=None,
+        *,
+        read_size=0,
+        use_list=True,
+        raw=False,
+        timestamp=0,
+        strict_map_key=True,
+        object_hook=None,
+        object_pairs_hook=None,
+        list_hook=None,
+        unicode_errors=None,
+        max_buffer_size=100 * 1024 * 1024,
+        ext_hook=None,
+        max_str_len=-1,
+        max_bin_len=-1,
+        max_array_len=-1,
+        max_map_len=-1,
+        max_ext_len=-1,
+        chunk_size=4 * 1024 * 1024,
+        verify_algorithm='crc32',
+        breakpoint_callback=None,
+    ):
+        super().__init__(
+            file_like,
+            read_size=read_size,
+            use_list=use_list,
+            raw=raw,
+            timestamp=timestamp,
+            strict_map_key=strict_map_key,
+            object_hook=object_hook,
+            object_pairs_hook=object_pairs_hook,
+            list_hook=list_hook,
+            unicode_errors=unicode_errors,
+            max_buffer_size=max_buffer_size,
+            ext_hook=ext_hook,
+            max_str_len=max_str_len,
+            max_bin_len=max_bin_len,
+            max_array_len=max_array_len,
+            max_map_len=max_map_len,
+            max_ext_len=max_ext_len,
+        )
+
+        # Validate chunk size (1MB to 64MB)
+        if not (1 * 1024 * 1024 <= chunk_size <= 64 * 1024 * 1024):
+            raise ValueError("chunk_size must be between 1MB and 64MB")
+        self.chunk_size = chunk_size
+
+        # Validate verify algorithm
+        if verify_algorithm not in ['crc32', 'md5', 'sha1']:
+            raise ValueError("verify_algorithm must be one of 'crc32', 'md5', 'sha1'")
+        self.verify_algorithm = verify_algorithm
+
+        # Calculate header length based on verify algorithm
+        if self.verify_algorithm == 'crc32':
+            self.header_length = 8
+        elif self.verify_algorithm == 'md5':
+            self.header_length = 20  # 4 (chunk size) + 16 (md5)
+        elif self.verify_algorithm == 'sha1':
+            self.header_length = 24  # 4 (chunk size) + 20 (sha1)
+
+        self.breakpoint_callback = breakpoint_callback
+
+        # Chunk tracking
+        self.current_chunk = bytearray()
+        self.current_chunk_number = 0
+        self.breakpoint = None
+
+    def _update_breakpoint(self):
+        """Update breakpoint information"""
+        breakpoint_info = {
+            'chunk_number': self.current_chunk_number,
+            'stream_offset': self._stream_offset,
+            'buffer_offset': self._buff_i,
+            'checkpoint': self._buf_checkpoint,
+        }
+        if self.breakpoint_callback:
+            self.breakpoint_callback(breakpoint_info)
+        self.breakpoint = breakpoint_info
+
+    def _verify_chunk(self, chunk, checksum):
+        """Verify chunk integrity"""
+        if self.verify_algorithm == 'crc32':
+            import zlib
+            calculated = zlib.crc32(chunk) & 0xffffffff
+            if calculated != checksum:
+                raise ChunkVerifyError(f"CRC32 check failed: expected 0x{checksum:08x}, got 0x{calculated:08x}")
+        elif self.verify_algorithm == 'md5':
+            calculated = hashlib.md5(chunk).digest()
+            if calculated != checksum:
+                raise ChunkVerifyError("MD5 check failed")
+        elif self.verify_algorithm == 'sha1':
+            calculated = hashlib.sha1(chunk).digest()
+            if calculated != checksum:
+                raise ChunkVerifyError("SHA1 check failed")
+
+    def feed(self, next_bytes):
+        """Feed data to the unpacker, handling chunked verification"""
+        if not self._feeding:
+            raise RuntimeError("feed() cannot be used with file_like")
+
+        # If we have a breakpoint, skip ahead
+        if self.breakpoint:
+            # For simplicity, we'll assume the breakpoint is at chunk boundary
+            # In a real implementation, we'd handle partial chunks
+            self.current_chunk_number = self.breakpoint['chunk_number']
+            self._stream_offset = self.breakpoint['stream_offset']
+            self._buff_i = self.breakpoint['buffer_offset']
+            self._buf_checkpoint = self.breakpoint['checkpoint']
+            self.breakpoint = None
+
+        # Combine current_chunk with new data
+        combined_data = self.current_chunk + next_bytes
+        view = memoryview(combined_data)
+        chunk_length = len(view)
+        offset = 0
+
+        while offset < chunk_length:
+            # Check if we have enough data for header
+            if offset + self.header_length > chunk_length:
+                # Not enough data for header
+                self.current_chunk = view[offset:]
+                break
+
+            # Read header
+            header = view[offset:offset+self.header_length]
+            chunk_size = struct.unpack_from('!I', header)[0]
+
+            # Validate chunk size
+            if chunk_size < 0 or chunk_size > self.chunk_size:
+                raise FormatError(f"Invalid chunk size: {chunk_size}")
+
+            # Check if we have enough data for full chunk
+            if offset + self.header_length + chunk_size > chunk_length:
+                # Not enough data for full chunk
+                self.current_chunk = view[offset:]
+                break
+
+            # Read chunk data
+            chunk_data = view[offset+self.header_length:offset+self.header_length+chunk_size]
+
+            # Read checksum from header
+            if self.verify_algorithm == 'crc32':
+                checksum = struct.unpack_from('!I', header, 4)[0]
+            elif self.verify_algorithm == 'md5':
+                checksum = header[4:20]
+            elif self.verify_algorithm == 'sha1':
+                checksum = header[4:24]
+
+            # Verify chunk
+            self._verify_chunk(chunk_data, checksum)
+
+            # Feed chunk to parent unpacker
+            super().feed(chunk_data)
+
+            # Update breakpoint
+            self._update_breakpoint()
+
+            # Move to next chunk
+            offset += self.header_length + chunk_size
+            self.current_chunk_number += 1
+
+        # If offset reached chunk_length, clear current_chunk
+        if offset == chunk_length:
+            self.current_chunk = bytearray()
+
+
+class ChunkedPacker(Packer):
+    """Packer that generates chunked data with verification"""
+
+    def __init__(
+        self,
+        chunk_size=4 * 1024 * 1024,
+        verify_algorithm='crc32',
+        *args,
+        **kwargs
+    ):
+        super().__init__(*args, **kwargs)
+
+        # Validate chunk size
+        if not (1 * 1024 * 1024 <= chunk_size <= 64 * 1024 * 1024):
+            raise ValueError("chunk_size must be between 1MB and 64MB")
+        self.chunk_size = chunk_size
+
+        # Validate verify algorithm
+        if verify_algorithm not in ['crc32', 'md5', 'sha1']:
+            raise ValueError("verify_algorithm must be one of 'crc32', 'md5', 'sha1'")
+        self.verify_algorithm = verify_algorithm
+
+        # Buffer for data
+        self.buffer = bytearray()
+
+    def _finalize_chunk(self):
+        """Finalize current chunk and return it"""
+        if not self.buffer:
+            return b''
+
+        chunk_data = bytes(self.buffer)
+
+        # Calculate checksum
+        if self.verify_algorithm == 'crc32':
+            import zlib
+            checksum = zlib.crc32(chunk_data) & 0xffffffff
+            # Header: chunk size (4 bytes), checksum (4 bytes)
+            header = struct.pack('!II', len(chunk_data), checksum)
+        elif self.verify_algorithm == 'md5':
+            checksum = hashlib.md5(chunk_data).digest()
+            # Header: chunk size (4 bytes), checksum (16 bytes)
+            header = struct.pack('!I', len(chunk_data)) + checksum
+        elif self.verify_algorithm == 'sha1':
+            checksum = hashlib.sha1(chunk_data).digest()
+            # Header: chunk size (4 bytes), checksum (20 bytes)
+            header = struct.pack('!I', len(chunk_data)) + checksum
+
+        self.buffer.clear()
+        return header + chunk_data
+
+    def pack(self, obj):
+        """Pack an object into chunked data"""
+        # Use super().pack() to get raw data
+        raw_data = super().pack(obj)
+
+        # Split into chunks
+        for i in range(0, len(raw_data), self.chunk_size):
+            chunk = raw_data[i:i+self.chunk_size]
+            self.buffer.extend(chunk)
+            if len(self.buffer) >= self.chunk_size:
+                yield self._finalize_chunk()
+
+        # Finalize any remaining data
+        if self.buffer:
+            yield self._finalize_chunk()
+
+    def packb(self, obj):
+        """Pack an object into a single chunk (if fits) or multiple chunks"""
+        return b''.join(self.pack(obj))
diff --git a/msgpack/exceptions.py b/msgpack/exceptions.py
@@ -42,6 +42,14 @@ def __str__(self):
         return "unpack(b) received extra data."
 
 
+class ChunkVerifyError(UnpackException):
+    """Chunk verification failed"""
+
+
+class BreakpointCorruptError(UnpackException):
+    """Breakpoint data is corrupt"""
+
+
 # Deprecated.  Use Exception instead to catch all exception during packing.
 PackException = Exception
 PackValueError = ValueError

diff --git a/msgpack/test_import.py b/msgpack/test_import.py
@@ -0,0 +1,10 @@
+#!/usr/bin/env python
+
+import traceback
+
+try:
+    import chunked
+    print('Import successful!')
+except Exception as e:
+    print(f'Import failed: {e}')
+    traceback.print_exc()