Add skipper history/manifest/platform commands

 - Add StreamTemplate and Operations methods for
   history/manifest/platform commands
 - Add hack to fix plaintext retrieval as a response from Skipper server
 - Make explicit Skipper client configuration with StringHttpMessageConverter taking the highest precedence

Resolves #1804
Resolves #1805
Resolves #1803

Author: ilayaperumalg

spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/StreamOperations.java

@@ -16,10 +16,13 @@
 
 package org.springframework.cloud.dataflow.rest.client;
 
+import java.util.Collection;
 import java.util.Map;
 
 import org.springframework.cloud.dataflow.rest.resource.StreamDefinitionResource;
+import org.springframework.cloud.skipper.domain.Deployer;
 import org.springframework.cloud.skipper.domain.PackageIdentifier;
+import org.springframework.cloud.skipper.domain.Release;
 import org.springframework.hateoas.PagedResources;
 
 /**
@@ -105,4 +108,25 @@ void updateStream(String streamName, String releaseName, PackageIdentifier packa
 	 */
 	StreamDefinitionResource getStreamDefinition(String streamName);
 
+	/**
+	 * Get manifest for the given stream deployed via Skipper.
+	 * Optionally, the version can be used to retrieve the version for a specific version of the stream.
+	 * @param streamName the stream(release) name
+	 * @param version the version of the release
+	 * @return the manifest for the given stream and version
+	 */
+	String getManifest(String streamName, int version);
+
+	/**
+	 * Get the history of releases for the given stream deployed via Skipper.
+	 * @param streamName the stream(release) name
+	 * @param max the maximum number of revisions to include in the history
+	 * @return the history of releases for the stream
+	 */
+	Collection<Release> history(String streamName, int max);
+
+	/**
+	 * @return the list of all Skipper platforms
+	 */
+	Collection<Deployer> listPlatforms();
 }

spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/StreamTemplate.java

@@ -16,13 +16,19 @@
 
 package org.springframework.cloud.dataflow.rest.client;
 
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.Map;
 
 import org.springframework.cloud.dataflow.rest.UpdateStreamRequest;
 import org.springframework.cloud.dataflow.rest.resource.StreamDefinitionResource;
+import org.springframework.cloud.skipper.domain.Deployer;
 import org.springframework.cloud.skipper.domain.PackageIdentifier;
+import org.springframework.cloud.skipper.domain.Release;
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.hateoas.Link;
 import org.springframework.hateoas.ResourceSupport;
+import org.springframework.http.HttpMethod;
 import org.springframework.util.Assert;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
@@ -132,6 +138,54 @@ public void rollbackStream(String streamName, int version) {
 		restTemplate.postForObject(url, null, Object.class);
 	}
 
+
+	@Override
+	public String getManifest(String streamName, int version) {
+		Assert.hasText(streamName, "Release name cannot be null or empty");
+		String url = String.format("%s/%s/%s/%s", deploymentsLink.getHref(), "manifest", streamName, version);
+		String manifest = restTemplate.getForObject(url, String.class);
+		// TODO - DataFlow only uses Jackson Marshaller, which does strange things to Strings as return values.
+		// \n is converted to two ascii characters 92 and 110...
+		String prunedManifest = manifest.substring(1, manifest.length() - 1);
+		StringBuilder sb = new StringBuilder();
+		for (int i = 0; i < prunedManifest.length(); i++) {
+			char testChar = prunedManifest.charAt(i);
+			if ( (int)testChar == 92) {
+				if (i != prunedManifest.length() -1) {
+					char nChar = prunedManifest.charAt(i+1);
+					if ( (int)nChar == 110) {
+						sb.append("\n");
+						i++;
+					}
+				}
+			} else {
+				sb.append(testChar);
+			}
+		}
+		return sb.toString();
+	}
+
+	@Override
+	public Collection<Release> history(String streamName, int maxRevisions) {
+		Assert.hasText(streamName, "Release name cannot be null or empty");
+		ParameterizedTypeReference<Collection<Release>> typeReference = new ParameterizedTypeReference<Collection<Release>>
+				() {
+		};
+		Map<String, Object> parameters = new HashMap<>();
+		String url = String.format("%s/%s/%s/%s", deploymentsLink.getHref(), "history", streamName, maxRevisions);
+		return this.restTemplate.exchange(url, HttpMethod.GET, null, typeReference, parameters).getBody();
+	}
+
+	@Override
+	public Collection<Deployer> listPlatforms() {
+		ParameterizedTypeReference<Collection<Deployer>> typeReference = new ParameterizedTypeReference<Collection<Deployer>>
+				() {
+		};
+		Map<String, Object> parameters = new HashMap<>();
+		String url = url = deploymentsLink.getHref() + "/platform/list";
+		return this.restTemplate.exchange(url, HttpMethod.GET, null, typeReference, parameters).getBody();
+	}
+
 	@Override
 	public StreamDefinitionResource getStreamDefinition(String streamName) {
 		String uriTemplate = this.definitionLink.expand(streamName).getHref();

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/config/DataFlowControllerAutoConfiguration.java

@@ -16,11 +16,13 @@
 
 package org.springframework.cloud.dataflow.server.config;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 
 import javax.sql.DataSource;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -36,6 +38,7 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.web.client.RestTemplateBuilder;
 import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
 import org.springframework.cloud.common.security.AuthorizationProperties;
 import org.springframework.cloud.common.security.support.FileSecurityProperties;
@@ -89,17 +92,21 @@
 import org.springframework.cloud.deployer.resource.support.DelegatingResourceLoader;
 import org.springframework.cloud.deployer.spi.app.AppDeployer;
 import org.springframework.cloud.deployer.spi.task.TaskLauncher;
+import org.springframework.cloud.skipper.client.DefaultSkipperClient;
 import org.springframework.cloud.skipper.client.SkipperClient;
-import org.springframework.cloud.skipper.client.SkipperClientConfiguration;
 import org.springframework.cloud.skipper.client.SkipperClientProperties;
+import org.springframework.cloud.skipper.client.SkipperClientResponseErrorHandler;
 import org.springframework.cloud.task.repository.TaskExplorer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.core.io.ResourceLoader;
 import org.springframework.hateoas.EntityLinks;
+import org.springframework.http.converter.StringHttpMessageConverter;
+import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
 import org.springframework.scheduling.concurrent.ForkJoinPoolFactoryBean;
+import org.springframework.web.client.RestTemplate;
 
 /**
  * Configuration for the Data Flow Server Controllers.
@@ -156,8 +163,8 @@ public StreamDefinitionController streamDefinitionController(StreamDefinitionRep
 	@Bean
 	@ConditionalOnBean({ StreamDefinitionRepository.class, StreamDeploymentRepository.class })
 	public StreamDeploymentController streamDeploymentController(StreamDefinitionRepository repository,
-			StreamService streamService) {
-		return new StreamDeploymentController(repository, streamService);
+			StreamService streamService, SkipperClient skipperClient) {
+		return new StreamDeploymentController(repository, streamService, skipperClient);
 	}
 
 	@Bean
@@ -378,10 +385,19 @@ public SecurityStateBean securityStateBean() {
 	}
 
 	@Configuration
-	@Import(SkipperClientConfiguration.class)
 	@ConditionalOnBean({ StreamDefinitionRepository.class, StreamDeploymentRepository.class })
+	@EnableConfigurationProperties(SkipperClientProperties.class)
 	public static class SkipperConfiguration {
 
+		@Bean
+		public SkipperClient skipperClient(SkipperClientProperties properties,
+				RestTemplateBuilder restTemplateBuilder, ObjectMapper objectMapper) {
+			RestTemplate restTemplate = restTemplateBuilder.errorHandler(new SkipperClientResponseErrorHandler
+					(objectMapper)).messageConverters(Arrays.asList(new StringHttpMessageConverter(), new
+					MappingJackson2HttpMessageConverter(objectMapper))).build();
+			return new DefaultSkipperClient(properties.getUri(), restTemplate);
+		}
+
 		@Bean
 		public SkipperStreamDeployer skipperStreamDeployer(SkipperClient skipperClient,
 				StreamDeploymentRepository streamDeploymentRepository,

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/StreamDeploymentController.java

@@ -16,15 +16,22 @@
 
 package org.springframework.cloud.dataflow.server.controller;
 
+import java.util.Collection;
 import java.util.Map;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.springframework.cloud.dataflow.core.StreamDefinition;
 import org.springframework.cloud.dataflow.rest.UpdateStreamRequest;
 import org.springframework.cloud.dataflow.rest.resource.StreamDeploymentResource;
 import org.springframework.cloud.dataflow.server.repository.NoSuchStreamDefinitionException;
 import org.springframework.cloud.dataflow.server.repository.StreamDefinitionRepository;
 import org.springframework.cloud.dataflow.server.service.StreamService;
 import org.springframework.cloud.deployer.spi.app.AppDeployer;
+import org.springframework.cloud.skipper.client.SkipperClient;
+import org.springframework.cloud.skipper.domain.Deployer;
+import org.springframework.cloud.skipper.domain.Release;
 import org.springframework.hateoas.ExposesResourceFor;
 import org.springframework.http.HttpStatus;
 import org.springframework.util.Assert;
@@ -50,8 +57,12 @@
 @ExposesResourceFor(StreamDeploymentResource.class)
 public class StreamDeploymentController {
 
+	private static final Logger logger = LoggerFactory.getLogger(StreamDeploymentController.class);
+
 	private final StreamService streamService;
 
+	private final SkipperClient skipperClient;
+
 	/**
 	 * The repository this controller will use for stream CRUD operations.
 	 */
@@ -67,12 +78,16 @@ public class StreamDeploymentController {
 	 *
 	 * @param repository the repository this controller will use for stream CRUD operations
 	 * @param streamService the underlying StreamService to deploy the stream
+	 * @param skipperClient the Skipper client to use for Skipper related services
 	 */
-	public StreamDeploymentController(StreamDefinitionRepository repository, StreamService streamService) {
+	public StreamDeploymentController(StreamDefinitionRepository repository, StreamService streamService,
+			SkipperClient skipperClient) {
 		Assert.notNull(repository, "StreamDefinitionRepository must not be null");
 		Assert.notNull(streamService, "StreamService must not be null");
+		Assert.notNull(skipperClient, "SkipperClient must not be null");
 		this.repository = repository;
 		this.streamService = streamService;
+		this.skipperClient = skipperClient;
 	}
 
 	/**
@@ -126,4 +141,32 @@ public void rollback(@PathVariable("name") String name, @PathVariable("version")
 		this.streamService.rollbackStream(name, version);
 	}
 
+	@RequestMapping(value = "/manifest/{name}/{version}", method = RequestMethod.GET)
+	@ResponseStatus(HttpStatus.OK)
+	public String manifest(@PathVariable("name") String name, @PathVariable("version") int version) {
+		if (version > 0) {
+			return this.skipperClient.manifest(name, version);
+		}
+		else {
+			return this.skipperClient.manifest(name);
+		}
+	}
+
+	@RequestMapping(path = "/history/{name}/{max}", method = RequestMethod.GET)
+	@ResponseStatus(HttpStatus.OK)
+	public Collection<Release> history(@PathVariable("name") String releaseName,
+			@PathVariable("max") int maxRevisions) {
+		if (maxRevisions > 0) {
+			return this.skipperClient.history(releaseName, String.valueOf(maxRevisions));
+		}
+		else {
+			return this.skipperClient.history(releaseName).getContent();
+		}
+	}
+
+	@RequestMapping(path = "/platform/list", method = RequestMethod.GET)
+	@ResponseStatus(HttpStatus.OK)
+	public Collection<Deployer> platformList() {
+		return this.skipperClient.listDeployers().getContent();
+	}
 }

spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/configuration/TestDependencies.java

@@ -133,8 +133,8 @@ public ResourceLoader resourceLoader() {
 
 	@Bean
 	public StreamDeploymentController streamDeploymentController(StreamDefinitionRepository repository,
-			StreamService streamService) {
-		return new StreamDeploymentController(repository, streamService);
+			StreamService streamService, SkipperClient skipperClient) {
+		return new StreamDeploymentController(repository, streamService, skipperClient);
 	}
 
 	@Bean

spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/StreamControllerTests.java

@@ -52,6 +52,7 @@
 import org.springframework.cloud.deployer.spi.app.AppStatus;
 import org.springframework.cloud.deployer.spi.app.DeploymentState;
 import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
+import org.springframework.cloud.skipper.client.SkipperClient;
 import org.springframework.http.MediaType;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.annotation.DirtiesContext.ClassMode;
@@ -125,6 +126,9 @@ public class StreamControllerTests {
 	@Autowired
 	private DefaultStreamService defaultStreamService;
 
+	@Autowired
+	private SkipperClient skipperClient;
+
 	@Before
 	public void setupMocks() {
 		this.mockMvc = MockMvcBuilders.webAppContextSetup(wac)
@@ -141,7 +145,7 @@ public void tearDown() {
 	@Test(expected = IllegalArgumentException.class)
 	public void testConstructorMissingRepository() {
 		StreamDeploymentController deploymentController = new StreamDeploymentController(
-				new InMemoryStreamDefinitionRepository(), defaultStreamService);
+				new InMemoryStreamDefinitionRepository(), defaultStreamService, skipperClient);
 		new StreamDefinitionController(null, appRegistry, defaultStreamService);
 	}
 

spring-cloud-dataflow-server-core/src/test/java/org/springframework/cloud/dataflow/server/controller/StreamDeploymentControllerTests.java

@@ -16,6 +16,10 @@
 
 package org.springframework.cloud.dataflow.server.controller;
 
+import static org.mockito.Mockito.*;
+import static org.springframework.cloud.dataflow.rest.SkipperStream.SKIPPER_ENABLED_PROPERTY_KEY;
+
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -28,12 +32,10 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
-
 import org.springframework.cloud.dataflow.server.repository.StreamDefinitionRepository;
 import org.springframework.cloud.dataflow.server.service.impl.DefaultStreamService;
-
-import static org.mockito.Mockito.verify;
-import static org.springframework.cloud.dataflow.rest.SkipperStream.SKIPPER_ENABLED_PROPERTY_KEY;
+import org.springframework.cloud.skipper.client.SkipperClient;
+import org.springframework.hateoas.Resources;
 
 /**
  * Unit tests for StreamDeploymentController.
@@ -55,9 +57,12 @@ public class StreamDeploymentControllerTests {
 	@Mock
 	private DefaultStreamService defaultStreamService;
 
+	@Mock
+	private SkipperClient skipperClient;
+
 	@Before
 	public void setup() {
-		this.controller = new StreamDeploymentController(streamDefinitionRepository, defaultStreamService);
+		this.controller = new StreamDeploymentController(streamDefinitionRepository, defaultStreamService, skipperClient);
 	}
 
 	@Test
@@ -84,4 +89,11 @@ public void testRollbackViaStreamService() {
 		Assert.assertTrue("Rollback version is incorrect", argumentCaptor2.getValue() == 2);
 	}
 
+	@Test
+	public void tesPlatformsListViaSkipperClient() {
+		when(skipperClient.listDeployers()).thenReturn(new Resources<>(new ArrayList<>(), new ArrayList<>()));
+		this.controller.platformList();
+		verify(skipperClient, times(1)).listDeployers();
+	}
+
 }