How to avoid try with Future fromTry

Question

I need to write a future that flushes a stream and closes it. Here is what I tried:

def close(ous: OutputStream) = Future.fromTry(Try {
    try {
      ous.flush()
    } finally {
      ous.close()
    }
  })

Looks ugly. try{}finally{} inside the Try. But I need finally block in order to avoid resource leak. Is there a way to rewrite the code in not so ugly way?

Tried StreamConverters? https://doc.akka.io/api/akka/current/akka/stream/scaladsl/StreamConverters$.html — Viktor Klang, May 22 '18 at 13:45

score 1 · Answer 1 · answered May 23 '18 at 03:29

Future already captures exceptions returning a Future.failed, there's no need to do fromTry and block, so you can do:

Future { out.flush() }.andThen( _=> out.close )

( Future { out.flush() } will asynchronously flush the stream and andThen will get called wether it finishes or fails.

score 0 · Answer 2 · answered May 21 '18 at 19:09

0

This could be an option i think.

def close(ous: OutputStream) = Future.fromTry(Try(ous.flush())) andThen {
    case Success(_) => println("do something here")
    case Failure(_) => ous.close()
}

answered May 21 '18 at 19:09

Hayk Hakobyan

309
2
8

1

it should close on the Success too (it's a finally) ? – Andy Hayden May 21 '18 at 21:11

prayagupa · Answer 3 · 2018-05-21T22:27:25.013

since you are already using Try use pattern match on result to Try{ stream.flush } and apply Try{ stream.close() }

example,

  import java.io.{ByteArrayOutputStream, OutputStream}
  import java.util.Date
  import java.io.ObjectOutputStream
  import scala.concurrent.Future
  import scala.util.Try
  import scala.util.{Failure, Success}
  import scala.concurrent.ExecutionContext.Implicits.global

  def doSomeOperation: OutputStream => Future[String] = (outputStream: OutputStream) =>
    withCleanup(outputStream) {
      Future {
        //1/0
        outputStream.toString
      }
    }

  def withCleanup(outputStream: OutputStream)(fn: Future[String]): Future[String] = {

    val execution = fn

    execution onComplete {
      case Success(_) => cleanup(outputStream)
      case Failure(_) => cleanup(outputStream)
    }

    execution
  }

  def cleanup(outputStream: OutputStream): Try[Unit] = Try {
    outputStream.flush()
    println("flushed")
  } match {
    case _ => Try { 
      outputStream.close()
      println("closed")
    }
  }

Then call the function, which will flush and close your stream as well.

val stream = new ObjectOutputStream(new ByteArrayOutputStream())    
stream.writeObject(new Date())

scala> doSomeOperation(stream)
res18: scala.concurrent.Future[String] = Future(<not completed>)
flushed
closed

Doesn't doing map here mean if there is an earlier failure it won't close? — Andy Hayden, May 21 '18 at 21:10
@AndyHayden good catch. updated to attempt closing stream even when main execution or `_.flush()` fails — prayagupa, May 21 '18 at 22:29

score 0 · Answer 4 · answered May 21 '18 at 21:35

It's not clear to me whether this is actually cleaner:

def close(ous: OutputStream) = Future.fromTry(
  val flushed = Try { ous.flush() }
  val closed = Try { ous.close() }
  if (closed.isFailure) closed else flushed  // bubble up the correct error
)

Note: This is nearly an equivalent of this answer but not quite. Mainly since the .close can fail, and this must be encapsulated in the Try.

How to avoid try with Future fromTry

4 Answers4