Play2 Framework 프록시 스트리밍 콘텐츠가 클라이언트에 스트리밍이 끝난 후 연결을 유지합니다.

Question

아래 코드는 클라이언트로 스트리밍됩니다. 내가 수집 한 것은 Java의 IO 스트림을 사용하는 것보다 더 관용적 인 방법입니다. 그러나 스트림에는 스트림이 완료된 후에도 연결이 유지됩니다.

def getImage() = Action { request => 
    val imageUrl = "http://hereandthere.com/someimageurl.png" 
    Ok.stream({ content: Iteratee[Array[Byte], Unit] => 
    WS.url(imageUrl).withHeaders("Accept"->"image/png").get { response => content } 
    return 
    }).withHeaders("Content-Type"->"image/png") 
} 

이것은 내부 API에서 요청자로 큰 파일 (> 1MB)을 스트리밍하기위한 것입니다.

질문은 왜 연결을 유지합니까? 업스트림 서버에서 기대할 수있는 것이 있습니까? 나는 업스트림 서버를 컬 (curl)을 사용하여 테스트했으며 연결은 닫히지 않습니다.이 프록시를 통과 할 때 닫히지 않습니다.

Answer 1

스트림이 완료되지 않는 이유는 WS.get() 호출에서 돌아 오는 iteratee로 EOF가 전송되지 않기 때문입니다. 이 명시 적 EOF가 없으면 연결은 청크 모드로 열려 있고, 잠재적으로 장기간 실행되는 혜성 같은 연결로 열려 있습니다.

다음

Ok.stream({ content: Iteratee[Array[Byte], Unit] => 
    WS.url(imageUrl) 
    .withHeaders("Accept"->"image/png") 
    .get { response => content } 
    .onRedeem { ii => 
     ii.feed(Input.EOF) 
    } 
}).withHeaders("Content-Type"->"image/png") 

Answer 2

플레이 2.1.0의 수정 된 버전입니다 :

여기에 고정 코드입니다. https://groups.google.com/forum/#!msg/play-framework/HwoRR-nipCc/gUKs9NexCx4J

공유를위한 Anatoly G를 참조하십시오. 플레이 2.2.x에 대한

def proxy = Action { 

    val url = "..." 

    Async { 
    val iterateePromise = Promise[Iteratee[Array[Byte], Unit]] 
    val resultPromise = Promise[ChunkedResult[Array[Byte]]] 

    WS.url(url).get { responseHeaders => 
     resultPromise.success { 
     new Status(responseHeaders.status).stream({ content: Iteratee[Array[Byte], Unit] => 
      iterateePromise.success(content) 
     }).withHeaders(
      "Content-Type" -> responseHeaders.headers.getOrElse("Content-Type", Seq("application/octet-stream")).head, 
      "Connection" -> "Close") 
     } 
     Iteratee.flatten(iterateePromise.future) 
    }.onComplete { 
     case Success(ii) => ii.feed(Input.EOF) 
     case Failure(t) => resultPromise.failure(t) 
    } 

    resultPromise.future 
    } 

} 

Answer 3

업데이트 :

def proxy = Action.async { 
    val url = "http://localhost:9000" 

    def enumerator(chunks: Iteratee[Array[Byte], Unit] => _) = { 
    new Enumerator[Array[Byte]] { 
     def apply[C](i: Iteratee[Array[Byte], C]): Future[Iteratee[Array[Byte], C]] = { 
     val doneIteratee = Promise[Iteratee[Array[Byte], C]]() 
     chunks(i.map { 
      done => 
      doneIteratee.success(Done[Array[Byte], C](done)).asInstanceOf[Unit] 
     }) 
     doneIteratee.future 
     } 
    } 
    } 

    val iterateePromise = Promise[Iteratee[Array[Byte], Unit]]() 
    val resultPromise = Promise[SimpleResult]() 

    WS.url(url).get { 
    responseHeaders => 

     resultPromise.success(new Status(responseHeaders.status).chunked(
     enumerator({ 
      content: Iteratee[Array[Byte], Unit] => iterateePromise.success(content) 
     } 
     )).withHeaders(
     "Content-Type" -> responseHeaders.headers.getOrElse("Content-Type", Seq("application/octet-stream")).head, 
     "Connection" -> "Close")) 

     Iteratee.flatten(iterateePromise.future) 
    }.onComplete { 
    case Success(ii) => ii.feed(Input.EOF) 
    case Failure(t) => throw t 
    } 

    resultPromise.future 
} 

사람이 더 나은 솔루션이있는 경우, 그것은 나를 크게 관심!