code.grnet.gr Git - aquarium/blob - src/main/scala/gr/grnet/aquarium/message/avro/AvroHelpers.scala

   1 /*
   2  * Copyright 2011-2012 GRNET S.A. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or
   5  * without modification, are permitted provided that the following
   6  * conditions are met:
   7  *
   8  *   1. Redistributions of source code must retain the above
   9  *      copyright notice, this list of conditions and the following
  10  *      disclaimer.
  11  *
  12  *   2. Redistributions in binary form must reproduce the above
  13  *      copyright notice, this list of conditions and the following
  14  *      disclaimer in the documentation and/or other materials
  15  *      provided with the distribution.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  18  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  19  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  20  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  21  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  24  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  25  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  27  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  28  * POSSIBILITY OF SUCH DAMAGE.
  29  *
  30  * The views and conclusions contained in the software and
  31  * documentation are those of the authors and should not be
  32  * interpreted as representing official policies, either expressed
  33  * or implied, of GRNET S.A.
  34  */
  35
  36 package gr.grnet.aquarium.message.avro
  37
  38 import gr.grnet.aquarium.util.json.JsonHelpers
  39 import java.io.{OutputStream, ByteArrayOutputStream}
  40 import org.apache.avro.Schema
  41 import org.apache.avro.generic.{GenericRecord, GenericDatumWriter}
  42 import org.apache.avro.io.{JsonDecoder, DecoderFactory, JsonEncoder, EncoderFactory}
  43 import org.apache.avro.specific.{SpecificDatumReader, SpecificDatumWriter, SpecificRecord}
  44
  45
  46 /**
  47  * Provides helper methods for generic Avro-related facilities.
  48  *
  49  * @author Christos KK Loverdos <loverdos@gmail.com>
  50  */
  51 object AvroHelpers {
  52   private[this] final val DefaultEncoderFactory = EncoderFactory.get()
  53   private[this] final val DefaultDecoderFactory = DecoderFactory.get()
  54
  55   def getJsonEncoder(schema: Schema, out: OutputStream): JsonEncoder = {
  56     val encoder = DefaultEncoderFactory.jsonEncoder(schema, out)
  57     val jsonGenerator = JsonHelpers.getJsonGenerator(out)
  58     encoder.configure(jsonGenerator)
  59   }
  60
  61   def getJsonDecoder(schema: Schema, in: String): JsonDecoder = {
  62     DefaultDecoderFactory.jsonDecoder(schema, in)
  63   }
  64
  65   def specificRecordOfJsonString[R <: SpecificRecord](json: String, fresh: R): R = {
  66     val schema = fresh.getSchema
  67     val decoder = getJsonDecoder(schema, json)
  68     val reader = new SpecificDatumReader[R](schema)
  69     reader.read(fresh, decoder)
  70   }
  71
  72   def jsonStringOfSpecificRecord[T <: SpecificRecord](t: T): String = {
  73     val schema = t.getSchema()
  74     val out = new ByteArrayOutputStream()
  75     val encoder = getJsonEncoder(schema, out)
  76     val writer = new SpecificDatumWriter[T](schema)
  77
  78     writer.write(t, encoder)
  79     encoder.flush()
  80     out.toString
  81   }
  82
  83   def jsonStringOfGenericRecord[T <: GenericRecord](t: T): String = {
  84     val schema = t.getSchema()
  85     val out = new ByteArrayOutputStream()
  86     val encoder = getJsonEncoder(schema, out)
  87     val writer = new GenericDatumWriter[T](schema)
  88
  89     writer.write(t, encoder)
  90     encoder.flush()
  91     out.toString
  92   }
  93
  94   def bytesOfSpecificRecord[R <: SpecificRecord](r: R): Array[Byte] = {
  95     val schema = r.getSchema
  96     val out = new ByteArrayOutputStream()
  97     val encoder = EncoderFactory.get().binaryEncoder(out, null)
  98     val writer = new SpecificDatumWriter[R](schema)
  99
 100     writer.write(r, encoder)
 101     encoder.flush()
 102     out.toByteArray
 103   }
 104
 105   def specificRecordOfBytes[R <: SpecificRecord](bytes: Array[Byte], fresh: R): R = {
 106     val decoder = DefaultDecoderFactory.binaryDecoder(bytes, null)
 107     val reader = new SpecificDatumReader[R](fresh.getSchema)
 108     reader.read(fresh, decoder)
 109   }
 110 }