archive.rs

pub struct Envelope<M: Message> {
	message: M,
	context: SpanContext,
}

impl<M: Message> Envelope<M> {
	pub fn new(message: M) -> Self {
		let context = Span::current().context().span().span_context().clone();
		Self { message, context }
	}
}

#[repr(C)]
pub struct ArchivedEnvelope {
	type_id: rkyv::Archived<u64>,
	type_name: rkyv::Archived<Box<str>>,
	message: rkyv::Archived<Box<[u8]>>,
	context: rkyv::Archived<[u8; 29]>,
}

pub struct EnvelopeResolver {
	type_name: rkyv::Resolver<Box<str>>,
	message: rkyv::boxed::BoxResolver<rkyv::rend::LittleEndian<u32>>,
	context: rkyv::Resolver<[u8; 29]>,
	context_data: [u8; 29],
}

impl<M: Message> rkyv::Archive for Envelope<M> {
	type Archived = ArchivedEnvelope;
	type Resolver = EnvelopeResolver;

	unsafe fn resolve(&self, pos: usize, resolver: Self::Resolver, out: *mut Self::Archived) {
		let (fp, fo) = rkyv::out_field!(out.type_id);
		rkyv::Archive::resolve(&M::ID.0, pos + fp, (), fo);

		let (fp, fo) = rkyv::out_field!(out.type_name);
		rkyv::boxed::ArchivedBox::resolve_from_ref(M::NAME, pos + fp, resolver.type_name, fo);

		let (fp, fo) = rkyv::out_field!(out.message);
		rkyv::boxed::ArchivedBox::resolve_from_raw_parts(pos + fp, resolver.message, fo);

		let (fp, fo) = rkyv::out_field!(out.context);
		rkyv::Archive::resolve(&resolver.context_data, pos + fp, resolver.context, fo);
	}
}

impl<M: Message, S: Serializer + ?Sized> rkyv::Serialize<S> for Envelope<M>
where
	M: Serialize<S>,
{
	fn serialize(&self, serializer: &mut S) -> Result<Self::Resolver, S::Error> {
		let type_name_resolver = rkyv::boxed::ArchivedBox::serialize_from_ref(M::NAME, serializer)?;

		let message_start = serializer.pos();
		serializer.serialize_value(&self.message)?;
		let message_end = serializer.pos();

		let message_size = u32::try_from(message_end - message_start).expect("msg too large");
		let message_size = rkyv::rend::LittleEndian::<u32>::new(message_size);

		let message_resolver =
			unsafe { rkyv::boxed::BoxResolver::from_raw_parts(message_start, message_size) };

		let context = opentelemetry_contrib::trace::propagator::binary::BinaryPropagator::new()
			.serialize_into_bytes(&self.context);
		let context_resolver = context.serialize(serializer)?;

		Ok(EnvelopeResolver {
			type_name: type_name_resolver,
			message: message_resolver,
			context: context_resolver,
			context_data: context,
		})
	}
}