package com.ctrip.hermes.meta.resource;

import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import javax.inject.Singleton;
import javax.ws.rs.DELETE;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
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.codehaus.plexus.util.StringUtils;
import org.unidal.dal.jdbc.DalException;
import org.unidal.dal.jdbc.DalNotFoundException;

import com.alibaba.fastjson.JSON;
import com.ctrip.hermes.core.utils.PlexusComponentLocator;
import com.ctrip.hermes.meta.entity.Storage;
import com.ctrip.hermes.meta.entity.Topic;
import com.ctrip.hermes.meta.pojo.SchemaView;
import com.ctrip.hermes.meta.pojo.TopicView;
import com.ctrip.hermes.meta.server.RestException;
import com.ctrip.hermes.meta.service.SchemaService;
import com.ctrip.hermes.meta.service.TopicService;

@Path("/topics/")
@Singleton
@Produces(MediaType.APPLICATION_JSON)
public class TopicResource {

	private static TopicService topicService = PlexusComponentLocator.lookup(TopicService.class);

	private static SchemaService schemaService = PlexusComponentLocator.lookup(SchemaService.class);

	@POST
	public Response createTopic(String content) {
		if (StringUtils.isEmpty(content)) {
			throw new RestException("HTTP POST body is empty", Status.BAD_REQUEST);
		}
		TopicView topicView = null;
		try {
			topicView = JSON.parseObject(content, TopicView.class);
		} catch (Exception e) {
			throw new RestException(e, Status.BAD_REQUEST);
		}

		Topic topic = topicView.toMetaTopic();

		if (topicService.getTopic(topic.getName()) != null) {
			throw new RestException("Topic already exists.", Status.CONFLICT);
		}
		try {
			topic = topicService.createTopic(topic);
			topicView = new TopicView(topic);
		} catch (Exception e) {
			throw new RestException(e, Status.INTERNAL_SERVER_ERROR);
		}
		return Response.status(Status.CREATED).entity(topicView).build();
	}

	@GET
	public List<TopicView> findTopics(@QueryParam("pattern") String pattern) {
		if (StringUtils.isEmpty(pattern)) {
			pattern = ".*";
		}

		List<Topic> topics = topicService.findTopics(pattern);
		List<TopicView> returnResult = new ArrayList<TopicView>();
		try {
			for (Topic topic : topics) {
				TopicView topicView = new TopicView(topic);
				Storage storage = topicService.findStorage(topic.getName());
				topicView.setStorage(storage);
				if (topic.getSchemaId() != null && topic.getSchemaId() > 0) {
					try {
						SchemaView schemaView = schemaService.getSchemaView(topic.getSchemaId());
						topicView.setSchema(schemaView);
					} catch (DalNotFoundException e) {
					}
				}
				returnResult.add(topicView);
			}
		} catch (DalException | IOException | RestClientException e) {
			throw new RestException(e, Status.INTERNAL_SERVER_ERROR);
		}
		return returnResult;
	}

	@GET
	@Path("{name}")
	public TopicView getTopic(@PathParam("name") String name) {
		Topic topic = topicService.getTopic(name);
		if (topic == null) {
			throw new RestException("Topic not found: " + name, Status.NOT_FOUND);
		}

		TopicView topicView = new TopicView(topic);
		if (topic.getSchemaId() > 0) {
			SchemaView schemaView;
			try {
				schemaView = schemaService.getSchemaView(topic.getSchemaId());
			} catch (DalException | IOException | RestClientException e) {
				throw new RestException(e, Status.INTERNAL_SERVER_ERROR);
			}
			topicView.setSchema(schemaView);
		}

		return topicView;
	}

	@PUT
	@Path("{name}")
	public Response updateTopic(@PathParam("name") String name, String content) {
		if (StringUtils.isEmpty(content)) {
			throw new RestException("HTTP PUT body is empty", Status.BAD_REQUEST);
		}
		TopicView topicView = null;
		try {
			topicView = JSON.parseObject(content, TopicView.class);
			topicView.setName(name);
		} catch (Exception e) {
			throw new RestException(e, Status.BAD_REQUEST);
		}

		Topic topic = topicView.toMetaTopic();

		if (topicService.getTopic(topic.getName()) == null) {
			throw new RestException("Topic does not exists.", Status.NOT_FOUND);
		}
		try {
			topic = topicService.updateTopic(topic);
			topicView = new TopicView(topic);
		} catch (Exception e) {
			throw new RestException(e, Status.INTERNAL_SERVER_ERROR);
		}
		return Response.status(Status.CREATED).entity(topicView).build();
	}

	@DELETE
	@Path("{name}")
	public Response deleteTopic(@PathParam("name") String name) {
		try {
			topicService.deleteTopic(name);
		} catch (Exception e) {
			throw new RestException(e, Status.INTERNAL_SERVER_ERROR);
		}
		return Response.status(Status.OK).build();
	}
}

@@ -2675,35 +2675,8 @@
 null
- && topic.getSchemaId() %3E 0
 ) %7B%0A
@@ -3330,16 +3330,112 @@
 topic);%0A
+%09%09Storage storage = topicService.findStorage(topic.getName());%0A%09%09topicView.setStorage(storage);%0A
 %09%09if (to
@@ -3456,11 +3456,15 @@
 d() 
-%3E 0
+!= null
 ) %7B%0A
@@ -3565,16 +3565,53 @@
 aId());%0A
+%09%09%09%09topicView.setSchema(schemaView);%0A
 %09%09%09%7D cat
@@ -3734,49 +3734,13 @@
 R);%0A
+
 %09%09%09%7D%0A
-%09%09%09topicView.setSchema(schemaView);%0A
 %09%09%7D%0A

Fix Schema getSchemaId autobox issue