Skip to content

A
atlas
Commit c1995379 by Harish Butani
Options

add support for createInstance/getInstance to simpleserver

parent 003b5a52
Showing with 198 additions and 40 deletions
typesystem/src/main/java/org/apache/hadoop/metadata/storage/DiscoverInstances.java
......@@ -64,13 +64,15 @@ public class DiscoverInstances implements ObjectGraphWalker.NodeProcessor {
if ( !idToNewIdMap.containsKey(id)) {
idToNewIdMap.put(id, repository.newId(id.className));
}
if ( idToInstanceMap.containsKey(ref)) {
if ( ref != null && idToInstanceMap.containsKey(ref)) {
// Oops
throw new RepositoryException(
String.format("Unexpected internal error: Id %s processed again", id));
}
if ( ref != null ) {
idToInstanceMap.put(id, ref);
}
}
}
}
}
typesystem/src/main/java/org/apache/hadoop/metadata/types/ClassType.java
......@@ -81,7 +81,7 @@ public class ClassType extends HierarchicalType<ClassType, IReferenceableInstanc
if ( val != null ) {
if ( val instanceof ITypedReferenceableInstance) {
ITypedReferenceableInstance tr = (ITypedReferenceableInstance) val;
if ( tr.getTypeName() != getName() ) {
if ( !tr.getTypeName().equals(getName()) ) {
/*
* If val is a subType instance; invoke convert on it.
*/
......
typesystem/src/main/scala/org/apache/hadoop/metadata/json/Serialization.scala
......@@ -18,18 +18,15 @@
package org.apache.hadoop.metadata.json
import org.apache.jute.compiler.JLong
import org.apache.hadoop.metadata.types.DataTypes.{MapType, TypeCategory, ArrayType}
import org.apache.hadoop.metadata._
import org.apache.hadoop.metadata.storage.{Id, ReferenceableInstance}
import org.apache.hadoop.metadata.types.DataTypes.{ArrayType, MapType, TypeCategory}
import org.apache.hadoop.metadata.types._
import org.apache.hadoop.metadata.storage.{ReferenceableInstance, Id}
import org.json4s.JsonAST.JInt
import org.json4s._
import org.json4s.native.Serialization.{write => swrite, _}
import org.json4s.reflect.{ScalaType, Reflector}
import java.util.regex.Pattern
import java.util.Date
import collection.JavaConversions._
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
class BigDecimalSerializer extends CustomSerializer[java.math.BigDecimal](format => ( {
......@@ -50,6 +47,15 @@ class IdSerializer extends CustomSerializer[Id](format => ( {
case JObject(JField("id", JInt(id)) ::
JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
case JObject(JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
JField("id", JInt(id)) ::
JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
case JObject(JField("id", JString(id)) ::
JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
JField("version", JString(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
case JObject(JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
JField("id", JString(id)) ::
JField("version", JString(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
}, {
case id: Id => JObject(JField("id", JInt(id.id)),
JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(id.className)),
......@@ -57,17 +63,19 @@ class IdSerializer extends CustomSerializer[Id](format => ( {
}
))
class TypedStructSerializer extends Serializer[ITypedStruct] {
class TypedStructSerializer(val typSystem : Option[MetadataService] = None) extends Serializer[ITypedStruct] {
def currentMdSvc = typSystem.getOrElse(MetadataService.getCurrentService())
def deserialize(implicit format: Formats) = {
case (TypeInfo(clazz, ptype), json) if classOf[ITypedStruct].isAssignableFrom(clazz) => json match {
case JObject(fs) =>
val(typ, fields) = fs.partition(f => f._1 == Serialization.STRUCT_TYPE_FIELD_NAME)
val typName = typ(0)._2.asInstanceOf[JString].s
val sT = MetadataService.getCurrentTypeSystem().getDataType(
val sT = currentMdSvc.getTypeSystem.getDataType(
classOf[IConstructableType[IStruct, ITypedStruct]], typName).asInstanceOf[IConstructableType[IStruct, ITypedStruct]]
val s = sT.createInstance()
Serialization.deserializeFields(sT, s, fields)
Serialization.deserializeFields(currentMdSvc, sT, s, fields)
s
case x => throw new MappingException("Can't convert " + x + " to TypedStruct")
}
......@@ -88,7 +96,10 @@ class TypedStructSerializer extends Serializer[ITypedStruct] {
}
}
class TypedReferenceableInstanceSerializer extends Serializer[ITypedReferenceableInstance] {
class TypedReferenceableInstanceSerializer(val typSystem : Option[MetadataService] = None)
extends Serializer[ITypedReferenceableInstance] {
def currentMdSvc = typSystem.getOrElse(MetadataService.getCurrentService())
def deserialize(implicit format: Formats) = {
case (TypeInfo(clazz, ptype), json) if classOf[ITypedReferenceableInstance].isAssignableFrom(clazz) => json match {
......@@ -120,11 +131,11 @@ class TypedReferenceableInstanceSerializer extends Serializer[ITypedReferenceabl
}
val typName = typField.get._2.asInstanceOf[JString].s
val sT = MetadataService.getCurrentTypeSystem().getDataType(
val sT = currentMdSvc.getTypeSystem.getDataType(
classOf[ClassType], typName).asInstanceOf[ClassType]
val id = Serialization.deserializeId(idField.get._2)
val s = sT.createInstance(id, traitNames:_*)
Serialization.deserializeFields(sT, s, fields)
Serialization.deserializeFields(currentMdSvc, sT, s, fields)
traitsField.map { t =>
val tObj :JObject = t._2.asInstanceOf[JObject]
......@@ -132,10 +143,10 @@ class TypedReferenceableInstanceSerializer extends Serializer[ITypedReferenceabl
val tName : String = oTrait._1
val traitJObj : JObject = oTrait._2.asInstanceOf[JObject]
val traitObj = s.getTrait(tName).asInstanceOf[ITypedStruct]
val tT = MetadataService.getCurrentTypeSystem().getDataType(
val tT = currentMdSvc.getTypeSystem.getDataType(
classOf[TraitType], traitObj.getTypeName).asInstanceOf[TraitType]
val(tTyp, tFields) = traitJObj.obj.partition(f => f._1 == Serialization.STRUCT_TYPE_FIELD_NAME)
Serialization.deserializeFields(tT, traitObj, tFields)
Serialization.deserializeFields(currentMdSvc, tT, traitObj, tFields)
}
}
......@@ -217,9 +228,12 @@ object Serialization {
}
}.toList.map(_.asInstanceOf[JField])
def deserializeFields[T <: ITypedInstance](sT : IConstructableType[_, T],
def deserializeFields[T <: ITypedInstance](currentMdSvc: MetadataService,
sT : IConstructableType[_, T],
s : T, fields : List[JField] )(implicit format: Formats)
= fields.foreach { f =>
= {
MetadataService.setCurrentService(currentMdSvc)
fields.foreach { f =>
val fName = f._1
val fInfo = sT.fieldMapping.fields(fName)
if ( fInfo != null ) {
......@@ -236,6 +250,7 @@ object Serialization {
s.set(fName, Serialization.extract(fInfo.dataType(), v))
}
}
}
def deserializeId(value : JValue)(implicit format: Formats) = value match {
case JObject(JField("id", JInt(id)) ::
......
typesystem/src/main/scala/org/apache/hadoop/metadata/tools/simpleserver/Main.scala
......@@ -21,6 +21,8 @@ package org.apache.hadoop.metadata.tools.simpleserver
import akka.actor.{Props, ActorSystem}
import akka.io.IO
import com.typesafe.config.ConfigFactory
import org.apache.hadoop.metadata.storage.memory.MemRepository
import org.apache.hadoop.metadata.types.TypeSystem
import spray.can.Http
/**
......@@ -32,9 +34,14 @@ import spray.can.Http
* http GET localhost:9140/listTypeNames
* pbpaste | http PUT localhost:9140/defineTypes
* http GET localhost:9140/typeDetails typeNames:='["Department", "Person", "Manager"]'
*
* pbpaste | http PUT localhost:9140/createInstance
* pbpaste | http GET localhost:9140/getInstance
* }}}
*
* - On the Mac, pbpaste makes available what is copied to clipboard. Copy contents of resources/sampleTypes.json
* - for createInstance resources/sampleInstance.json is an example
* - for getInstance send an Id back, you can copy the output from createInstance.
*
*/
object Main extends App {
......@@ -44,6 +51,9 @@ object Main extends App {
implicit val system = ActorSystem("metadataservice")
val api = system.actorOf(Props(new RestInterface()), "httpInterface")
val typSys = new TypeSystem
val memRepo = new MemRepository(typSys)
val api = system.actorOf(Props(new RestInterface(typSys, memRepo)), "httpInterface")
IO(Http) ! Http.Bind(listener = api, interface = host, port = port)
}
\ No newline at end of file
typesystem/src/main/scala/org/apache/hadoop/metadata/tools/simpleserver/MetadataService.scala → typesystem/src/main/scala/org/apache/hadoop/metadata/tools/simpleserver/MetadataActor.scala
......@@ -21,41 +21,65 @@ package org.apache.hadoop.metadata.tools.simpleserver
import akka.actor._
import akka.util.Timeout
import com.google.common.collect.ImmutableList
import org.apache.hadoop.metadata.{MetadataService, ITypedReferenceableInstance}
import org.apache.hadoop.metadata.json._
import org.apache.hadoop.metadata.storage.memory.MemRepository
import org.apache.hadoop.metadata.types.{IDataType, TypeSystem}
import org.apache.hadoop.metadata.types._
import org.json4s.{Formats, NoTypeHints}
import spray.httpx.Json4sSupport
import org.apache.hadoop.metadata.storage.Id
import scala.concurrent.duration._
class MetadataService extends Actor with ActorLogging {
class MetadataActor(val typeSystem: TypeSystem, val memRepository : MemRepository) extends Actor with ActorLogging {
import org.apache.hadoop.metadata.tools.simpleserver.MetadataProtocol._
import scala.collection.JavaConversions._
import scala.language.postfixOps
implicit val timeout = Timeout(5 seconds)
val typSys = new TypeSystem
val memRepo = new MemRepository(typSys)
def receive = {
case ListTypeNames() =>
sender ! TypeNames(typSys.getTypeNames.toList)
sender ! TypeNames(typeSystem.getTypeNames.toList)
case GetTypeDetails(typeNames) =>
val typesDef = TypesSerialization.convertToTypesDef(typSys, (d : IDataType[_]) => typeNames.contains(d.getName))
val typesDef = TypesSerialization.convertToTypesDef(typeSystem, (d : IDataType[_]) => typeNames.contains(d.getName))
sender ! TypeDetails(typesDef)
case DefineTypes(typesDef : TypesDef) =>
typesDef.enumTypes.foreach(typSys.defineEnumType(_))
typesDef.enumTypes.foreach(typeSystem.defineEnumType(_))
typSys.defineTypes(ImmutableList.copyOf(typesDef.structTypes.toArray),
typeSystem.defineTypes(ImmutableList.copyOf(typesDef.structTypes.toArray),
ImmutableList.copyOf(typesDef.traitTypes.toArray),
ImmutableList.copyOf(typesDef.classTypes.toArray))
var newTypes : List[HierarchicalType[_ <: HierarchicalType[_ <: AnyRef, _], _]] = Nil
typesDef.traitTypes.foreach { tDef =>
val nm = tDef.typeName
newTypes = newTypes :+
typeSystem.getDataType(classOf[HierarchicalType[_ <: HierarchicalType[_ <: AnyRef, _], _]], nm)
}
typesDef.classTypes.foreach { tDef =>
val nm = tDef.typeName
newTypes = newTypes :+
typeSystem.getDataType(classOf[HierarchicalType[_ <: HierarchicalType[_ <: AnyRef, _], _]], nm)
}
memRepository.defineTypes(newTypes)
sender ! TypesCreated
case CreateInstance(i) =>
MetadataService.setCurrentService(new MetadataService(memRepository, typeSystem))
val r = memRepository.create(i)
sender ! InstanceCreated(r.getId)
case GetInstance(id) =>
MetadataService.setCurrentService(new MetadataService(memRepository, typeSystem))
val r = memRepository.get(id)
sender ! InstanceDetails(r)
}
}
......@@ -67,13 +91,22 @@ object MetadataProtocol {
case class TypeDetails(types : TypesDef)
case class DefineTypes(types : TypesDef)
case class TypesCreated()
case class CreateInstance(i: ITypedReferenceableInstance)
case class InstanceCreated(id : Id)
case class GetInstance(id : Id)
case class InstanceDetails(i: ITypedReferenceableInstance)
}
object Json4sProtocol extends Json4sSupport {
trait Json4sProtocol extends Json4sSupport {
val typeSystem : TypeSystem
val memRepository : MemRepository
implicit def json4sFormats: Formats =
org.json4s.native.Serialization.formats(NoTypeHints) + new MultiplicitySerializer + new TypedStructSerializer +
new TypedReferenceableInstanceSerializer + new BigDecimalSerializer + new BigIntegerSerializer
org.json4s.native.Serialization.formats(NoTypeHints) + new MultiplicitySerializer +
new TypedStructSerializer(Some(new MetadataService(memRepository, typeSystem))) +
new TypedReferenceableInstanceSerializer(Some(new MetadataService(memRepository, typeSystem))) +
new BigDecimalSerializer + new BigIntegerSerializer + new IdSerializer
}
typesystem/src/main/scala/org/apache/hadoop/metadata/tools/simpleserver/RestInterface.scala
......@@ -20,13 +20,17 @@ package org.apache.hadoop.metadata.tools.simpleserver
import akka.actor._
import akka.util.Timeout
import org.apache.hadoop.metadata.{MetadataService, ITypedReferenceableInstance}
import org.apache.hadoop.metadata.json.TypesDef
import org.apache.hadoop.metadata.storage.Id
import org.apache.hadoop.metadata.storage.memory.MemRepository
import org.apache.hadoop.metadata.types.TypeSystem
import spray.http.StatusCodes
import spray.routing._
import scala.concurrent.duration._
class Responder(requestContext:RequestContext, ticketMaster:ActorRef) extends Actor with ActorLogging {
import org.apache.hadoop.metadata.tools.simpleserver.Json4sProtocol._
class Responder(val typeSystem: TypeSystem, val memRepository : MemRepository,
requestContext:RequestContext, mdSvc:ActorRef) extends Actor with Json4sProtocol with ActorLogging {
import org.apache.hadoop.metadata.tools.simpleserver.MetadataProtocol._
def receive = {
......@@ -42,33 +46,40 @@ class Responder(requestContext:RequestContext, ticketMaster:ActorRef) extends Ac
case TypesCreated =>
requestContext.complete(StatusCodes.OK)
self ! PoisonPill
case InstanceCreated(id) =>
requestContext.complete(StatusCodes.OK, id)
case InstanceDetails(i) =>
requestContext.complete(StatusCodes.OK, i)
}
}
class RestInterface extends HttpServiceActor
class RestInterface(val typeSystem: TypeSystem, val memRepository : MemRepository) extends HttpServiceActor
with RestApi {
def receive = runRoute(routes)
}
trait RestApi extends HttpService with ActorLogging { actor: Actor =>
trait RestApi extends HttpService with Json4sProtocol with ActorLogging { actor: Actor =>
import MetadataProtocol._
import scala.language.postfixOps
import scala.concurrent.ExecutionContext.Implicits.global
val typeSystem : TypeSystem
val memRepository : MemRepository
implicit val timeout = Timeout(10 seconds)
import akka.pattern.{ask, pipe}
val mdSvc = context.actorOf(Props[MetadataService])
import Json4sProtocol._
val mdSvc = context.actorOf(Props(new MetadataActor(typeSystem, memRepository)))
def routes: Route =
path("listTypeNames") {
get { requestContext =>
val responder : ActorRef = createResponder(requestContext)
val responder: ActorRef = createResponder(requestContext)
pipe(mdSvc.ask(ListTypeNames))
......@@ -90,9 +101,26 @@ trait RestApi extends HttpService with ActorLogging { actor: Actor =>
mdSvc.ask(DefineTypes(typesDef)).pipeTo(responder)
}
}
} ~
path("createInstance") {
put {
entity(as[ITypedReferenceableInstance]) { i => requestContext =>
val responder = createResponder(requestContext)
mdSvc.ask(CreateInstance(i)).pipeTo(responder)
}
}
} ~
path("getInstance") {
get {
entity(as[Id]) { id => requestContext =>
val responder = createResponder(requestContext)
mdSvc.ask(GetInstance(id)).pipeTo(responder)
}
}
}
def createResponder(requestContext:RequestContext) = {
context.actorOf(Props(new Responder(requestContext, mdSvc)))
context.actorOf(Props(new Responder(typeSystem, memRepository, requestContext, mdSvc)))
}
}
\ No newline at end of file
typesystem/src/test/resources/sampleInstance.json 0 → 100644
{
"$typeName$":"Department",
"$id$":{
"id":-1420494283853484000,
"$typeName$":"Department",
"version":0
},
"employees":[{
"$typeName$":"Person",
"$id$":{
"id":-1420494283853508000,
"$typeName$":"Person",
"version":0
},
"manager":{
"id":-1420494283853511000,
"$typeName$":"Manager",
"version":0
},
"department":{
"id":-1420494283853484000,
"$typeName$":"Department",
"version":0
},
"name":"John"
},{
"$typeName$":"Manager",
"$id$":{
"id":-1420494283853511000,
"$typeName$":"Manager",
"version":0
},
"manager":null,
"subordinates":[{
"$typeName$":"Person",
"$id$":{
"id":-1420494283853508000,
"$typeName$":"Person",
"version":0
},
"manager":{
"id":-1420494283853511000,
"$typeName$":"Manager",
"version":0
},
"department":{
"id":-1420494283853484000,
"$typeName$":"Department",
"version":0
},
"name":"John"
}],
"department":{
"id":-1420494283853484000,
"$typeName$":"Department",
"version":0
},
"name":"Jane",
"$traits$":{
"SecurityClearance":{
"$typeName$":"SecurityClearance",
"level":1
}
}
}],
"name":"hr"
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment