This repository has been archived by the owner on Jun 7, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 292
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #579 from zalando/ARUHA-591-start-from-cursors
Aruha 591 start from cursors
- Loading branch information
Showing
30 changed files
with
1,305 additions
and
577 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
119 changes: 119 additions & 0 deletions
119
src/main/java/org/zalando/nakadi/controller/PostSubscriptionController.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,119 @@ | ||
| package org.zalando.nakadi.controller; | ||
|
|
||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.http.HttpStatus; | ||
| import org.springframework.http.ResponseEntity; | ||
| import org.springframework.validation.Errors; | ||
| import org.springframework.web.bind.annotation.ExceptionHandler; | ||
| import org.springframework.web.bind.annotation.RequestBody; | ||
| import org.springframework.web.bind.annotation.RequestMapping; | ||
| import org.springframework.web.bind.annotation.RequestMethod; | ||
| import org.springframework.web.bind.annotation.RestController; | ||
| import org.springframework.web.context.request.NativeWebRequest; | ||
| import org.springframework.web.util.UriComponents; | ||
| import org.zalando.nakadi.domain.Subscription; | ||
| import org.zalando.nakadi.domain.SubscriptionBase; | ||
| import org.zalando.nakadi.exceptions.runtime.DuplicatedSubscriptionException; | ||
| import org.zalando.nakadi.exceptions.runtime.InconsistentStateException; | ||
| import org.zalando.nakadi.exceptions.runtime.MyNakadiRuntimeException1; | ||
| import org.zalando.nakadi.exceptions.runtime.NoEventTypeException; | ||
| import org.zalando.nakadi.exceptions.runtime.NoSubscriptionException; | ||
| import org.zalando.nakadi.exceptions.runtime.TooManyPartitionsException; | ||
| import org.zalando.nakadi.exceptions.runtime.WrongInitialCursorsException; | ||
| import org.zalando.nakadi.plugin.api.ApplicationService; | ||
| import org.zalando.nakadi.problem.ValidationProblem; | ||
| import org.zalando.nakadi.security.Client; | ||
| import org.zalando.nakadi.service.subscription.SubscriptionService; | ||
| import org.zalando.nakadi.util.FeatureToggleService; | ||
| import org.zalando.problem.MoreStatus; | ||
| import org.zalando.problem.Problem; | ||
| import org.zalando.problem.spring.web.advice.Responses; | ||
|
|
||
| import javax.validation.Valid; | ||
| import javax.ws.rs.core.HttpHeaders; | ||
| import javax.ws.rs.core.Response; | ||
|
|
||
| import static org.springframework.http.HttpStatus.NOT_IMPLEMENTED; | ||
| import static org.springframework.http.HttpStatus.OK; | ||
| import static org.zalando.nakadi.util.FeatureToggleService.Feature.CHECK_OWNING_APPLICATION; | ||
| import static org.zalando.nakadi.util.FeatureToggleService.Feature.DISABLE_SUBSCRIPTION_CREATION; | ||
| import static org.zalando.nakadi.util.FeatureToggleService.Feature.HIGH_LEVEL_API; | ||
|
|
||
|
|
||
| @RestController | ||
| public class PostSubscriptionController { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(PostSubscriptionController.class); | ||
|
|
||
| private final FeatureToggleService featureToggleService; | ||
| private final ApplicationService applicationService; | ||
| private final SubscriptionService subscriptionService; | ||
|
|
||
| @Autowired | ||
| public PostSubscriptionController(final FeatureToggleService featureToggleService, | ||
| final ApplicationService applicationService, | ||
| final SubscriptionService subscriptionService) { | ||
| this.featureToggleService = featureToggleService; | ||
| this.applicationService = applicationService; | ||
| this.subscriptionService = subscriptionService; | ||
| } | ||
|
|
||
| @RequestMapping(value = "/subscriptions", method = RequestMethod.POST) | ||
| public ResponseEntity<?> createOrGetSubscription(@Valid @RequestBody final SubscriptionBase subscriptionBase, | ||
| final Errors errors, | ||
| final NativeWebRequest request, | ||
| final Client client) { | ||
| if (!featureToggleService.isFeatureEnabled(HIGH_LEVEL_API)) { | ||
| return new ResponseEntity<>(NOT_IMPLEMENTED); | ||
| } | ||
| if (errors.hasErrors()) { | ||
| return Responses.create(new ValidationProblem(errors), request); | ||
| } | ||
| if (featureToggleService.isFeatureEnabled(CHECK_OWNING_APPLICATION) | ||
| && !applicationService.exists(subscriptionBase.getOwningApplication())) { | ||
| return Responses.create(Problem.valueOf(MoreStatus.UNPROCESSABLE_ENTITY, | ||
| "owning_application doesn't exist"), request); | ||
| } | ||
|
|
||
| try { | ||
| return ok(subscriptionService.getExistingSubscription(subscriptionBase)); | ||
| } catch (final NoSubscriptionException e) { | ||
| if (featureToggleService.isFeatureEnabled(DISABLE_SUBSCRIPTION_CREATION)) { | ||
| return Responses.create(Response.Status.SERVICE_UNAVAILABLE, | ||
| "Subscription creation is temporarily unavailable", request); | ||
| } | ||
| try { | ||
| final Subscription subscription = subscriptionService.createSubscription(subscriptionBase, client); | ||
| return prepareLocationResponse(subscription); | ||
| } catch (final DuplicatedSubscriptionException ex) { | ||
| throw new InconsistentStateException("Unexpected problem occurred when creating subscription", ex); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private ResponseEntity<?> ok(final Subscription existingSubscription) { | ||
| final UriComponents location = subscriptionService.getSubscriptionUri(existingSubscription); | ||
| return ResponseEntity.status(OK).location(location.toUri()).body(existingSubscription); | ||
| } | ||
|
|
||
| private ResponseEntity<?> prepareLocationResponse(final Subscription subscription) { | ||
| final UriComponents location = subscriptionService.getSubscriptionUri(subscription); | ||
| return ResponseEntity.status(HttpStatus.CREATED) | ||
| .location(location.toUri()) | ||
| .header(HttpHeaders.CONTENT_LOCATION, location.toString()) | ||
| .body(subscription); | ||
| } | ||
|
|
||
| @ExceptionHandler({ | ||
| NoEventTypeException.class, | ||
| WrongInitialCursorsException.class, | ||
| TooManyPartitionsException.class}) | ||
| public ResponseEntity<Problem> handleUnprocessableSubscription(final MyNakadiRuntimeException1 exception, | ||
| final NativeWebRequest request) { | ||
| LOG.debug("Error occurred when working with subscriptions", exception); | ||
| return Responses.create(MoreStatus.UNPROCESSABLE_ENTITY, exception.getMessage(), request); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.