package com.ctrip.hermes.metaserver.rest.resource;

import io.confluent.kafka.schemaregistry.client.rest.RestService;

import java.util.Map;

import javax.inject.Singleton;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.ctrip.hermes.core.utils.PlexusComponentLocator;
import com.ctrip.hermes.core.utils.StringUtils;
import com.ctrip.hermes.meta.entity.Codec;
import com.ctrip.hermes.metaserver.meta.MetaHolder;
import com.ctrip.hermes.metaserver.rest.commons.RestException;

@Path("/schema/register")
@Singleton
@Produces(MediaType.APPLICATION_JSON)
public class SchemaRegisterResource {
	private static final Logger log = LoggerFactory.getLogger(SchemaRegisterResource.class);

	private MetaHolder m_metaHolder = PlexusComponentLocator.lookup(MetaHolder.class);

	@POST
	public Response registerSchema(String content) {
		@SuppressWarnings("unchecked")
		Map<String, String> m = JSON.parseObject(content, Map.class);
		if (StringUtils.isBlank(m.get("schema")) || StringUtils.isBlank(m.get("subject"))) {
			throw new RestException("Schema and subject is required.", Status.BAD_REQUEST);
		}
		Codec codec = m_metaHolder.getMeta().getCodecs().get("avro");
		if (codec != null) {
			try {
				RestService restService = new RestService(codec.findProperty("schema.registry.url").getValue());
				int id = restService.registerSchema(m.get("schema"), m.get("subject"));
				return Response.status(Status.OK).entity(id).build();
			} catch (Exception e) {
				throw new RestException("Register schema failed.", e);
			}
		}
		log.warn("Register schema failed. Codec avro not found.");
		return Response.status(Status.NOT_FOUND).build();
	}

	@GET
	public Response getSchemaById(@QueryParam("id") int id) {
		Codec codec = m_metaHolder.getMeta().getCodecs().get("avro");
		if (codec != null) {
			try {
				RestService restService = new RestService(codec.findProperty("schema.registry.url").getValue());
				return Response.status(Status.OK).entity(restService.getId(id).getSchemaString()).build();
			} catch (Exception e) {
				throw new RestException("Get schema string failed.", e);
			}
		}
		log.warn("Get schema string failed. Codec avro not found.");
		return Response.status(Status.NOT_FOUND).build();
	}
}

@@ -166,16 +166,49 @@
 gleton;%0A
+import javax.ws.rs.DefaultValue;%0A
 import j
@@ -444,16 +444,140 @@
 tatus;%0A%0A
+import org.apache.http.HttpResponse;%0Aimport org.apache.http.client.fluent.Request;%0Aimport org.apache.http.util.EntityUtils;%0A
 import o
@@ -1215,16 +1215,1033 @@
 lass);%0A%0A
+%09@GET%0A%09@Path(%22get%22)%0A%09public Response getLatestSchemaBySubject(@QueryParam(%22subject%22) String subject) %7B%0A%09%09if (StringUtils.isBlank(subject)) %7B%0A%09%09%09throw new RestException(%22Subject is required.%22, Status.BAD_REQUEST);%0A%09%09%7D%0A%09%09Codec codec = m_metaHolder.getMeta().getCodecs().get(%22avro%22);%0A%09%09if (codec != null) %7B%0A%09%09%09try %7B%0A%09%09%09%09String url = codec.findProperty(%22schema.registry.url%22).getValue();%0A%09%09%09%09if (StringUtils.isBlank(url)) %7B%0A%09%09%09%09%09throw new RestException(%22Can not find schema registry url in codec.%22, Status.INTERNAL_SERVER_ERROR);%0A%09%09%09%09%7D%0A%09%09%09%09url = String.format(%22%25s/subjects/%25s/versions/latest%22, url, subject);%0A%09%09%09%09HttpResponse response = Request.Get(url).execute().returnResponse();%0A%09%09%09%09String json = EntityUtils.toString(response.getEntity());%0A%09%09%09%09return Response.status(response.getStatusLine().getStatusCode()).entity(json).build();%0A%09%09%09%7D catch (Exception e) %7B%0A%09%09%09%09throw new RestException(%22Register schema failed.%22, e);%0A%09%09%09%7D%0A%09%09%7D%0A%09%09return Response.status(Status.NOT_FOUND).entity(%22No avro codec was found!%22).build();%0A%09%7D%0A%0A
 %09@POST%0A%09
@@ -3082,32 +3082,67 @@
 atus.NOT_FOUND).
+entity(%22No avro codec was found!%22).
 build();%0A%09%7D%0A%0A%09@G
@@ -3192,16 +3192,36 @@
 am(%22id%22)
+ @DefaultValue(%22-1%22)
  int id)
@@ -3718,16 +3718,51 @@
 _FOUND).
+entity(%22No avro codec was found!%22).
 build();

add find schema api for meta-server