| 48 | |
| 49 | impl Header { |
| 50 | pub fn from_reader<R: AvroRead>(reader: &mut R) -> Result<Header, AvroError> { |
| 51 | let meta_schema = Schema { |
| 52 | named: vec![], |
| 53 | indices: Default::default(), |
| 54 | top: SchemaPiece::Map(Box::new(SchemaPiece::Bytes.into())).into(), |
| 55 | }; |
| 56 | |
| 57 | let mut buf = [0u8; 4]; |
| 58 | reader.read_exact(&mut buf)?; |
| 59 | |
| 60 | if buf != [b'O', b'b', b'j', 1u8] { |
| 61 | return Err(AvroError::Decode(DecodeError::WrongHeaderMagic(buf))); |
| 62 | } |
| 63 | |
| 64 | if let Value::Map(meta) = decode(meta_schema.top_node(), reader)? { |
| 65 | // TODO: surface original parse schema errors instead of coalescing them here |
| 66 | let json = meta |
| 67 | .get("avro.schema") |
| 68 | .ok_or(AvroError::Decode(DecodeError::MissingAvroDotSchema)) |
| 69 | .and_then(|bytes| { |
| 70 | if let Value::Bytes(ref bytes) = *bytes { |
| 71 | from_slice(bytes.as_ref()).map_err(|e| { |
| 72 | AvroError::ParseSchema(ParseSchemaError::new(format!( |
| 73 | "unable to decode schema bytes: {}", |
| 74 | e |
| 75 | ))) |
| 76 | }) |
| 77 | } else { |
| 78 | unreachable!() |
| 79 | } |
| 80 | })?; |
| 81 | let writer_schema = Schema::parse(&json).map_err(|e| { |
| 82 | ParseSchemaError::new(format!("unable to parse json as avro schema: {}", e)) |
| 83 | })?; |
| 84 | |
| 85 | let codec = meta |
| 86 | .get("avro.codec") |
| 87 | .map(|val| match val { |
| 88 | Value::Bytes(bytes) => from_utf8(bytes.as_ref()) |
| 89 | .map_err(|_e| AvroError::Decode(DecodeError::CodecUtf8Error)) |
| 90 | .and_then(|codec| { |
| 91 | Codec::from_str(codec).map_err(|_| { |
| 92 | AvroError::Decode(DecodeError::UnrecognizedCodec(codec.to_string())) |
| 93 | }) |
| 94 | }), |
| 95 | _ => unreachable!(), |
| 96 | }) |
| 97 | .unwrap_or(Ok(Codec::Null))?; |
| 98 | |
| 99 | let mut marker = [0u8; 16]; |
| 100 | reader.read_exact(&mut marker)?; |
| 101 | |
| 102 | Ok(Header { |
| 103 | writer_schema, |
| 104 | marker, |
| 105 | codec, |
| 106 | }) |
| 107 | } else { |