/*
 * Copyright 2015 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.cloud.data.rest.client;

import java.util.Collections;
import java.util.Map;

import org.springframework.cloud.data.rest.resource.StreamDefinitionResource;
import org.springframework.cloud.data.rest.util.DeploymentPropertiesUtils;
import org.springframework.hateoas.UriTemplate;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestTemplate;

/**
 * Implementation for {@link StreamOperations}.
 *
 * @author Ilayaperumal Gopinathan
 * @author Mark Fisher
 */
public class StreamTemplate implements StreamOperations {

	private final RestTemplate restTemplate;

	private final UriTemplate definitionsPath;

	private final UriTemplate deploymentsPath;

	StreamTemplate(RestTemplate restTemplate, Map<String, UriTemplate> resources) {
		this.restTemplate = restTemplate;
		this.definitionsPath = resources.get("streams/definitions");
		this.deploymentsPath = resources.get("streams/deployments");
	}

	@Override
	public StreamDefinitionResource.Page list() {
		String uriTemplate = definitionsPath.toString();
		uriTemplate = uriTemplate + "?size=10000";
		return restTemplate.getForObject(uriTemplate, StreamDefinitionResource.Page.class);
	}

	@Override
	public StreamDefinitionResource createStream(String name, String definition, boolean deploy) {
		MultiValueMap<String, Object> values = new LinkedMultiValueMap<String, Object>();
		values.add("name", name);
		values.add("definition", definition);
		values.add("deploy", Boolean.toString(deploy));
		StreamDefinitionResource stream = restTemplate.postForObject(
				definitionsPath.expand(), values, StreamDefinitionResource.class);
		return stream;
	}

	@Override
	public void deploy(String name, Map<String, String> properties) {
		String uriTemplate = deploymentsPath.toString() + "/{name}";
		MultiValueMap<String, Object> values = new LinkedMultiValueMap<String, Object>();
		values.add("properties", DeploymentPropertiesUtils.format(properties));
		restTemplate.postForObject(uriTemplate, values, Object.class, name);
	}

	@Override
	public void undeploy(String name) {
		String uriTemplate = deploymentsPath.toString() + "/{name}";
		restTemplate.delete(uriTemplate, name);
	}

	@Override
	public void undeployAll() {
		restTemplate.delete(deploymentsPath.expand());
	}

	@Override
	public void destroy(String name) {
		String uriTemplate = definitionsPath.toString() + "/{name}";
		restTemplate.delete(uriTemplate, Collections.singletonMap("name", name));
	}

	@Override
	public void destroyAll() {
		restTemplate.delete(definitionsPath.expand());
	}
}

@@ -917,16 +917,56 @@
 mplate;%0A
+import org.springframework.util.Assert;%0A
 import o
@@ -1286,16 +1286,160 @@
 ions %7B%0A%0A
+%09private static final String DEFINITIONS_PATH = %22streams/definitions%22;%0A%0A%09private static final String DEPLOYMENTS_PATH = %22streams/deployments%22;%0A%0A
 %09private
@@ -1644,16 +1644,242 @@
 rces) %7B%0A
+%09%09Assert.notNull(resources, %22URI Resources can't be null%22);%0A%09%09Assert.notNull(resources.get(DEFINITIONS_PATH), %22Definitions path is required%22);%0A%09%09Assert.notNull(resources.get(DEPLOYMENTS_PATH), %22Deployments path is required%22);%0A
 %09%09this.r
@@ -1945,37 +1945,32 @@
 ces.get(
-%22streams/definitions%22
+DEFINITIONS_PATH
 );%0A%09%09thi
@@ -2003,37 +2003,32 @@
 ces.get(
-%22streams/deployments%22
+DEPLOYMENTS_PATH
 );%0A%09%7D%0A%0A%09

Add assertions to StreamTemplate