Rename graph stage classes used for mongodb gridfs

Author: jiangpengcheng

common/scala/src/main/scala/org/apache/openwhisk/core/database/mongodb/MongoDBArtifactStore.scala

@@ -434,7 +434,7 @@ class MongoDBArtifactStore[DocumentAbstraction <: DocumentSerializer](client: Mo
     val option = new GridFSUploadOptions().metadata(document)
 
     val uploadStream = gridFSBucket.openUploadStream(BsonString(s"$id/$name"), name, option)
-    val sink = AsyncStreamSink(uploadStream)
+    val sink = MongoDBAsyncStreamSink(uploadStream)
 
     val f = docStream
       .runWith(combinedSink(sink))
@@ -498,7 +498,7 @@ class MongoDBArtifactStore[DocumentAbstraction <: DocumentSerializer](client: Mo
     val downloadStream = gridFSBucket.openDownloadStream(BsonString(s"${doc.id.id}/$attachmentName"))
 
     def readStream(file: GridFSFile) = {
-      val source = AsyncStreamSource(downloadStream)
+      val source = MongoDBAsyncStreamSource(downloadStream)
       source
         .runWith(sink)
         .map { result =>

common/scala/src/main/scala/org/apache/openwhisk/core/database/mongodb/AsyncStreamSink.scala renamed to common/scala/src/main/scala/org/apache/openwhisk/core/database/mongodb/MongoDBAsyncStreamSink.scala

@@ -30,7 +30,7 @@ import org.mongodb.scala.gridfs.{AsyncOutputStream}
 import scala.concurrent.{ExecutionContext, Future, Promise}
 import scala.util.{Failure, Success, Try}
 
-class AsyncStreamSink(stream: AsyncOutputStream)(implicit ec: ExecutionContext)
+class MongoDBAsyncStreamSink(stream: AsyncOutputStream)(implicit ec: ExecutionContext)
     extends GraphStageWithMaterializedValue[SinkShape[ByteString], Future[IOResult]] {
   val in: Inlet[ByteString] = Inlet("AsyncStream.in")
 
@@ -115,8 +115,8 @@ class AsyncStreamSink(stream: AsyncOutputStream)(implicit ec: ExecutionContext)
   }
 }
 
-object AsyncStreamSink {
+object MongoDBAsyncStreamSink {
   def apply(stream: AsyncOutputStream)(implicit ec: ExecutionContext): Sink[ByteString, Future[IOResult]] = {
-    Sink.fromGraph(new AsyncStreamSink(stream))
+    Sink.fromGraph(new MongoDBAsyncStreamSink(stream))
   }
 }

common/scala/src/main/scala/org/apache/openwhisk/core/database/mongodb/AsyncStreamSource.scala renamed to common/scala/src/main/scala/org/apache/openwhisk/core/database/mongodb/MongoDBAsyncStreamSource.scala

@@ -40,7 +40,7 @@ import scala.util.Success
 import scala.util.Try
 import scala.util.Failure
 
-class AsyncStreamSource(stream: AsyncInputStream, chunkSize: Int)(implicit ec: ExecutionContext)
+class MongoDBAsyncStreamSource(stream: AsyncInputStream, chunkSize: Int)(implicit ec: ExecutionContext)
     extends GraphStageWithMaterializedValue[SourceShape[ByteString], Future[IOResult]] {
   require(chunkSize > 0, "chunkSize must be greater than 0")
   val out: Outlet[ByteString] = Outlet("AsyncStream.out")
@@ -96,9 +96,9 @@ class AsyncStreamSource(stream: AsyncInputStream, chunkSize: Int)(implicit ec: E
   }
 }
 
-object AsyncStreamSource {
+object MongoDBAsyncStreamSource {
   def apply(stream: AsyncInputStream, chunkSize: Int = 512 * 1024)(
     implicit ec: ExecutionContext): Source[ByteString, Future[IOResult]] = {
-    Source.fromGraph(new AsyncStreamSource(stream, chunkSize))
+    Source.fromGraph(new MongoDBAsyncStreamSource(stream, chunkSize))
   }
 }

tests/src/test/scala/org/apache/openwhisk/core/database/mongodb/AsyncStreamGraphTests.scala renamed to tests/src/test/scala/org/apache/openwhisk/core/database/mongodb/MongoDBAsyncStreamGraphTests.scala

@@ -37,7 +37,7 @@ import org.scalatestplus.mockito.MockitoSugar
 import scala.util.Random
 
 @RunWith(classOf[JUnitRunner])
-class AsyncStreamGraphTests
+class MongoDBAsyncStreamGraphTests
     extends FlatSpec
     with Matchers
     with ScalaFutures
@@ -47,13 +47,13 @@ class AsyncStreamGraphTests
 
   implicit val mat = ActorMaterializer()
 
-  behavior of "AsyncStreamSource"
+  behavior of "MongoDBAsyncStreamSource"
 
   it should "read all bytes" in {
     val bytes = randomBytes(4000)
     val asyncStream = AsyncStreamHelper.toAsyncInputStream(bytes)
 
-    val readStream = AsyncStreamSource(asyncStream, 42).runWith(StreamConverters.asInputStream())
+    val readStream = MongoDBAsyncStreamSource(asyncStream, 42).runWith(StreamConverters.asInputStream())
     val readBytes = IOUtils.toByteArray(readStream)
 
     bytes shouldBe readBytes
@@ -65,7 +65,7 @@ class AsyncStreamGraphTests
     val spiedStream = spy(inputStream)
     val asyncStream = AsyncStreamHelper.toAsyncInputStream(spiedStream)
 
-    val readStream = AsyncStreamSource(asyncStream, 42).runWith(StreamConverters.asInputStream())
+    val readStream = MongoDBAsyncStreamSource(asyncStream, 42).runWith(StreamConverters.asInputStream())
     val readBytes = IOUtils.toByteArray(readStream)
 
     bytes shouldBe readBytes
@@ -79,7 +79,7 @@ class AsyncStreamGraphTests
     doThrow(exception).when(inputStream).read(any())
     val asyncStream = AsyncStreamHelper.toAsyncInputStream(inputStream)
 
-    val (ioResult, p) = AsyncStreamSource(asyncStream).toMat(Sink.asPublisher(false))(Keep.both).run()
+    val (ioResult, p) = MongoDBAsyncStreamSource(asyncStream).toMat(Sink.asPublisher(false))(Keep.both).run()
     val c = TestSubscriber.manualProbe[ByteString]()
     p.subscribe(c)
 
@@ -92,7 +92,7 @@ class AsyncStreamGraphTests
     ioResult.futureValue.status.isFailure shouldBe true
   }
 
-  behavior of "AsyncStreamSink"
+  behavior of "MongoDBAsyncStreamSink"
 
   it should "write all bytes" in {
     val bytes = randomBytes(4000)
@@ -101,7 +101,7 @@ class AsyncStreamGraphTests
     val os = new ByteArrayOutputStream()
     val asyncStream = AsyncStreamHelper.toAsyncOutputStream(os)
 
-    val sink = AsyncStreamSink(asyncStream)
+    val sink = MongoDBAsyncStreamSink(asyncStream)
     val ioResult = source.toMat(sink)(Keep.right).run()
 
     ioResult.futureValue.count shouldBe bytes.length
@@ -117,7 +117,7 @@ class AsyncStreamGraphTests
     val outputStream = new CloseRecordingStream()
     val asyncStream = AsyncStreamHelper.toAsyncOutputStream(outputStream)
 
-    val sink = AsyncStreamSink(asyncStream)
+    val sink = MongoDBAsyncStreamSink(asyncStream)
     val ioResult = source.toMat(sink)(Keep.right).run()
 
     ioResult.futureValue.count shouldBe 4000
@@ -129,7 +129,7 @@ class AsyncStreamGraphTests
     val os = new ByteArrayOutputStream()
     val asyncStream = AsyncStreamHelper.toAsyncOutputStream(os)
 
-    val sink = AsyncStreamSink(asyncStream)
+    val sink = MongoDBAsyncStreamSink(asyncStream)
     val ioResult = Source(1 to 10)
       .map { n ⇒
         if (n == 7) throw new Error("bees!")