diff --git a/sources/mongo/snapshot.go b/sources/mongo/snapshot.go
index d1447d9c..8e124a7a 100644
--- a/sources/mongo/snapshot.go
+++ b/sources/mongo/snapshot.go
@@ -11,7 +11,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/options"
 )
 
-type collectionScanner struct {
+type snapshot struct {
 	db         *mongo.Database
 	cfg        config.MongoDB
 	collection config.Collection
@@ -21,19 +21,19 @@ type collectionScanner struct {
 	done   bool
 }
 
-func newSnapshotIterator(db *mongo.Database, collection config.Collection, cfg config.MongoDB) *collectionScanner {
-	return &collectionScanner{
+func newSnapshotIterator(db *mongo.Database, collection config.Collection, cfg config.MongoDB) *snapshot {
+	return &snapshot{
 		db:         db,
 		cfg:        cfg,
 		collection: collection,
 	}
 }
 
-func (c *collectionScanner) HasNext() bool {
+func (c *snapshot) HasNext() bool {
 	return !c.done
 }
 
-func (c *collectionScanner) Next() ([]lib.RawMessage, error) {
+func (c *snapshot) Next() ([]lib.RawMessage, error) {
 	if !c.HasNext() {
 		return nil, fmt.Errorf("no more rows to scan")
 	}
diff --git a/sources/mongo/streaming.go b/sources/mongo/streaming.go
new file mode 100644
index 00000000..022bfd9d
--- /dev/null
+++ b/sources/mongo/streaming.go
@@ -0,0 +1,3 @@
+package mongo
+
+type