package nakadi;

import com.google.common.collect.Lists;
import com.google.gson.reflect.TypeToken;
import java.lang.reflect.Type;
import java.util.List;
import java.util.concurrent.TimeUnit;

class EventTypeResourceReal implements EventTypeResource {

  private static final String PATH_EVENT_TYPES = "event-types";
  private static final String PATH_PARTITIONS = "partitions";
  private static final String APPLICATION_JSON = "application/json";
  private static final Type TYPE = new TypeToken<List<EventType>>() {
  }.getType();
  private static final Type TYPE_P = new TypeToken<List<Partition>>() {
  }.getType();

  private final NakadiClient client;
  private String scope;

  EventTypeResourceReal(NakadiClient client) {
    this.client = client;
  }

  @Override public EventTypeResource scope(String scope) {
    this.scope = scope;
    return this;
  }

  @Override public Response create(EventType eventType)
      throws AuthorizationException, ClientException, ServerException, InvalidException,
      RateLimitException, NakadiException {

    PolicyBackoff backoff = ExponentialBackoff.newBuilder()
        .initialInterval(1000, TimeUnit.MILLISECONDS)
        .maxAttempts(3)
        .maxInterval(6000, TimeUnit.MILLISECONDS)
        .build();

    ResourceOptions options = prepareOptions(TokenProvider.NAKADI_EVENT_TYPE_WRITE);
    return client.resourceProvider()
        .newResource()
        .policyBackoff(backoff)
        .requestThrowing(Resource.POST, collectionUri().buildString(), options, eventType);
  }

  @Override public Response update(EventType eventType)
      throws AuthorizationException, ClientException, ServerException, InvalidException,
      RateLimitException, NakadiException {
    String url = collectionUri().path(eventType.name()).buildString();
    ResourceOptions options = prepareOptions(TokenProvider.NAKADI_EVENT_TYPE_WRITE);
    return client.resourceProvider().newResource()
        .requestThrowing(Resource.PUT, url, options, eventType);
  }

  @Override public EventType findByName(String eventTypeName)
      throws AuthorizationException, ClientException, ServerException, InvalidException,
      RateLimitException, NakadiException {
    String url = collectionUri().path(eventTypeName).buildString();
    // filebug: no scope defined on this resource; work with NAKADI_EVENT_STREAM_READ for now
    ResourceOptions options = prepareOptions(TokenProvider.NAKADI_EVENT_STREAM_READ);
    return client.resourceProvider().newResource()
        .requestThrowing(Resource.GET, url, options, EventType.class);
  }

  @Override public Response delete(String eventTypeName)
      throws AuthorizationException, ClientException, ServerException, InvalidException,
      RateLimitException, NakadiException {
    String url = collectionUri().path(eventTypeName).buildString();
    ResourceOptions options = prepareOptions(TokenProvider.NAKADI_CONFIG_WRITE);
    return client.resourceProvider().newResource()
        .requestThrowing(Resource.DELETE, url, options);
  }

  @Override public EventTypeCollection list()
      throws AuthorizationException, ClientException, ServerException, InvalidException,
      RateLimitException, NakadiException {
    return loadPage(collectionUri().buildString());
  }

  @Override public PartitionCollection partitions(String eventTypeName)
      throws AuthorizationException, ClientException, ServerException, InvalidException,
      RateLimitException, NakadiException {
    return loadPartitionPage(
        collectionUri().path(eventTypeName).path(PATH_PARTITIONS).buildString());
  }

  @Override public Partition partition(String eventTypeName, String partitionId)
      throws AuthorizationException, ClientException, ServerException, InvalidException,
      RateLimitException, NakadiException {
    final String url =
        collectionUri().path(eventTypeName).path(PATH_PARTITIONS).path(partitionId).buildString();
    ResourceOptions options = prepareOptions(TokenProvider.NAKADI_EVENT_STREAM_READ);
    return client.resourceProvider().newResource()
        .requestThrowing(Resource.GET, url, options, Partition.class);
  }

  String applyScope(String fallbackScope) {
    return scope == null ? fallbackScope: scope;
  }

  EventTypeCollection loadPage(String url) {
    // filebug: no scope defined on this resource; work with NAKADI_EVENT_STREAM_READ for now
    ResourceOptions options = prepareOptions(TokenProvider.NAKADI_EVENT_STREAM_READ);
    Response response = client.resourceProvider().newResource()
        .requestThrowing(Resource.GET, url, options);

    List<EventType> collection =
        client.jsonSupport().fromJson(response.responseBody().asString(), TYPE);

    return new EventTypeCollection(collection, Lists.newArrayList(), this);
  }

  PartitionCollection loadPartitionPage(String url) {
    ResourceOptions options = prepareOptions(TokenProvider.NAKADI_EVENT_STREAM_READ);
    Response response = client.resourceProvider().newResource()
        .requestThrowing(Resource.GET, url, options);

    List<Partition> collection =
        client.jsonSupport().fromJson(response.responseBody().asString(), TYPE_P);

    return new PartitionCollection(collection, Lists.newArrayList(), this);
  }

  private ResourceOptions prepareOptions(String fallbackScope) {
    return ResourceSupport.options(APPLICATION_JSON)
        .tokenProvider(client.resourceTokenProvider())
        .scope(applyScope(fallbackScope)); // use the set scope or fallback
  }

  private UriBuilder collectionUri() {
    return UriBuilder.builder(client.baseURI()).path(PATH_EVENT_TYPES);
  }
}

@@ -4025,32 +4025,41 @@
 sourceProvider()
+%0A        
 .newResource()%0A 
@@ -4049,32 +4049,89 @@
  .newResource()%0A
+        .policyBackoff(policyBackoffForCollectionPage())%0A
         .request
@@ -4554,32 +4554,41 @@
 sourceProvider()
+%0A        
 .newResource()%0A 
@@ -4578,32 +4578,89 @@
  .newResource()%0A
+        .policyBackoff(policyBackoffForCollectionPage())%0A
         .request
@@ -5075,32 +5075,41 @@
 sourceProvider()
+%0A        
 .newResource()%0A 
@@ -5099,32 +5099,89 @@
  .newResource()%0A
+        .policyBackoff(policyBackoffForCollectionPage())%0A
         .request
@@ -5706,24 +5706,24 @@
 tionUri() %7B%0A
-
     return U
@@ -5785,11 +5785,264 @@
 ES);%0A  %7D
+%0A%0A  private PolicyBackoff policyBackoffForCollectionPage() %7B%0A    return ExponentialBackoff.newBuilder()%0A        .initialInterval(1000, TimeUnit.MILLISECONDS)%0A        .maxAttempts(5)%0A        .maxInterval(6000, TimeUnit.MILLISECONDS)%0A        .build();%0A  %7D
 %0A%7D%0A

Add backoffs to event type calls