def to_rpc_shared_memory()

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

• 30 lines of code
• 4 McCabe index (conditional complexity)

    def to_rpc_shared_memory(
            cls,
            datum: 'Datum',
            shmem_mgr) -> Optional[protos.RpcSharedMemory]:
        """
        Writes the given value to shared memory and returns the corresponding
        RpcSharedMemory object which can be sent back to the functions host over
        RPC.
        """
        if datum.type == 'bytes':
            value = datum.value
            shared_mem_meta = shmem_mgr.put_bytes(value)
            data_type = protos.RpcDataType.bytes
        elif datum.type == 'string':
            value = datum.value
            shared_mem_meta = shmem_mgr.put_string(value)
            data_type = protos.RpcDataType.string
        else:
            raise NotImplementedError(
                f'Unsupported datum type ({datum.type}) for shared memory'
            )

        if shared_mem_meta is None:
            logger.warning('Cannot write to shared memory for type: %s',
                           datum.type)
            return None

        shmem = protos.RpcSharedMemory(
            name=shared_mem_meta.mem_map_name,
            offset=0,
            count=shared_mem_meta.count_bytes,
            type=data_type)

        logger.info(
            'Wrote %s bytes to memory map %s for data type %s',
            shared_mem_meta.count_bytes, shared_mem_meta.mem_map_name,
            data_type)
        return shmem