def from_rpc_shared_memory()

in azure_functions_worker/bindings/datumdef.py [0:0]

• 27 lines of code
• 7 McCabe index (conditional complexity)

    def from_rpc_shared_memory(
            cls,
            shmem: protos.RpcSharedMemory,
            shmem_mgr) -> Optional['Datum']:
        """
        Reads the specified shared memory region and converts the read data
        into a datum object of the corresponding type.
        """
        if shmem is None:
            logger.warning('Cannot read from shared memory. '
                           'RpcSharedMemory is None.')
            return None

        mem_map_name = shmem.name
        offset = shmem.offset
        count = shmem.count
        data_type = shmem.type
        ret_val = None

        if data_type == protos.RpcDataType.bytes:
            val = shmem_mgr.get_bytes(mem_map_name, offset, count)
            if val is not None:
                ret_val = cls(val, 'bytes')
        elif data_type == protos.RpcDataType.string:
            val = shmem_mgr.get_string(mem_map_name, offset, count)
            if val is not None:
                ret_val = cls(val, 'string')

        if ret_val is not None:
            logger.info(
                'Read %s bytes from memory map %s for data type %s', count,
                mem_map_name, data_type)
            return ret_val
        return None