package com.twitter.finatra.http.tests.streaming;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

import scala.Option;
import scala.collection.JavaConverters;

import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;

import com.twitter.concurrent.AsyncStream;
import com.twitter.finagle.http.Response;
import com.twitter.finagle.stats.StatsReceiver;
import com.twitter.finatra.http.internal.marshalling.MessageBodyManager;
import com.twitter.finatra.http.response.ResponseBuilder;
import com.twitter.finatra.http.streaming.StreamingResponse;
import com.twitter.finatra.http.streaming.ToReader;
import com.twitter.finatra.json.FinatraObjectMapper;
import com.twitter.finatra.utils.FileResolver;

import com.twitter.io.Buf;
import com.twitter.io.Bufs;
import com.twitter.io.Reader;
import com.twitter.io.Readers;
import com.twitter.io.StreamTermination;
import com.twitter.util.Await;
import com.twitter.util.Function;
import com.twitter.util.Future;

public class StreamingResponseJavaTest extends Assert {

  private Future burnLoop(Reader<Buf> reader) {
    return reader.read().flatMap(new Function<Option<Buf>, Future<scala.runtime.BoxedUnit>>() {
      @Override
      public Future<scala.runtime.BoxedUnit> apply(Option<Buf> v1) {
        if (v1.nonEmpty()) {
          return burnLoop(reader);
        } else {
          return Future.Unit();
        }
      }
    });
  }

  private ResponseBuilder responseBuilder = new ResponseBuilder(
    FinatraObjectMapper.create(null),
    new FileResolver("src/main/webapp/", ""),
    Mockito.mock(MessageBodyManager.class),
    Mockito.mock(StatsReceiver.class),
    true
  );

  private Response fromReader(Reader reader) throws Exception {
    StreamingResponse streamingResponse = responseBuilder.streaming(
      reader, ToReader.ReaderIdentity());
    Future<Response> fResponse = streamingResponse.toFutureResponse();
    return Await.result(fResponse);
  }

  private Reader<Buf> infiniteReader(Buf buf) {
    Stream<Buf> infiniteStream = Stream.iterate(buf, i -> i.concat(buf));
    return Readers.fromSeq(infiniteStream);
  }

  @Test
  public void emptyReader() throws Exception {
    Reader reader = Readers.newEmptyReader();
    Response response = fromReader(reader);
    // observe the EOS
    response.reader().read();
    Assert.assertEquals(
      Await.result(response.reader().onClose()), StreamTermination.FullyRead$.MODULE$);
  }

  @Test
  public void serdeReaderofString() throws Exception {
    List<String> stringList = Arrays.asList("first", "second", "third");
    Reader<String> reader = Readers.fromSeq(stringList);
    Response response = fromReader(reader);
    Future<Buf> fBuf = Readers.readAll(response.reader());
    String result = Buf.decodeString(Await.result(fBuf), StandardCharsets.UTF_8);
    Assert.assertEquals("\"first\"\"second\"\"third\"", result);
  }

  @Test
  public void serdeReaderofObject() throws Exception {
    FooClass f1 = new FooClass(1, "first");
    FooClass f2 = new FooClass(2, "second");
    List<FooClass> fooList = Arrays.asList(f1, f2);
    Reader<FooClass> reader = Readers.fromSeq(fooList);
    Response response = fromReader(reader);
    Future<Buf> fBuf = Readers.readAll(response.reader());
    String result = Buf.decodeString(Await.result(fBuf), StandardCharsets.UTF_8);
    Assert.assertEquals("{\"v1\":1,\"v2\":\"first\"}{\"v1\":2,\"v2\":\"second\"}", result);
  }

  @Test
  public void readerWriteFailureWithReaderDicardedException() throws Exception {
    Reader<Buf> reader = infiniteReader(Bufs.UTF_8.apply("foo"));
    Response response = fromReader(reader);
    response.reader().discard();
    burnLoop(response.reader());
    Assert.assertEquals(
      Await.result(response.reader().onClose()), StreamTermination.Discarded$.MODULE$);
  }

  private Response fromStream(AsyncStream stream) throws Exception {
    StreamingResponse streamingResponse = responseBuilder.streaming(
      stream, ToReader.AsyncStreamToReader());
    Future<Response> fResponse = streamingResponse.toFutureResponse();
    return Await.result(fResponse);
  }

  @Test
  public void serdeAsyncStreamOfString() throws Exception {
    List<String> stringList = Arrays.asList("first", "second", "third");
    AsyncStream<String> stream = AsyncStream.fromSeq(
      JavaConverters.asScalaIteratorConverter(stringList.iterator()).asScala().toSeq());
    Response response = fromStream(stream);
    Future<Buf> fBuf = Readers.readAll(response.reader());
    String result = Buf.decodeString(Await.result(fBuf), StandardCharsets.UTF_8);
    Assert.assertEquals("\"first\"\"second\"\"third\"", result);
  }

}

@@ -171,29 +171,8 @@
 m;%0A%0A
-import scala.Option;%0A
 impo
@@ -1160,20 +1160,16 @@
 Map(
-new 
 Function
 %3COpt
@@ -1168,146 +1168,22 @@
 tion
-%3COption%3CBuf%3E, Future%3Cscala.runtime.BoxedUnit%3E%3E() %7B%0A      @Override%0A      public Future%3Cscala.runtime.BoxedUnit%3E apply(Option%3CBuf%3E v1)
+.func(v1 -%3E
  %7B%0A
-  
     
@@ -1205,18 +1205,16 @@
 ty()) %7B%0A
-  
         
@@ -1244,18 +1244,16 @@
 ;%0A      
-  
 %7D else %7B
@@ -1253,18 +1253,16 @@
  else %7B%0A
-  
         
@@ -1287,26 +1287,16 @@
 ();%0A
-        %7D%0A
       %7D%0A
     
@@ -1295,14 +1295,15 @@
   %7D%0A
-
     %7D)
+)
 ;%0A  

Use lambdas in Java test