Checkpoint.

config/mongodb.go

@@ -9,12 +9,12 @@ import (
 )
 
 type MongoDB struct {
-	Host     string `yaml:"host"`
-	Username string `yaml:"username"`
-	Password string `yaml:"password"`
-	Database string `yaml:"database"`
-
+	Host        string       `yaml:"host"`
+	Username    string       `yaml:"username"`
+	Password    string       `yaml:"password"`
+	Database    string       `yaml:"database"`
 	Collections []Collection `yaml:"collections"`
+	Streaming   bool         `yaml:"streaming,omitempty"`
 }
 
 type Collection struct {

sources/mongo/mongo.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"github.com/artie-labs/reader/sources"
 	"log/slog"
 	"time"
 
@@ -15,12 +16,12 @@ import (
 	"github.com/artie-labs/reader/writers"
 )
 
-type Source struct {
+type SnapshotStore struct {
 	cfg config.MongoDB
 	db  *mongo.Database
 }
 
-func Load(cfg config.MongoDB) (*Source, error) {
+func Load(cfg config.MongoDB) (sources.Source, error) {
 	creds := options.Credential{
 		Username: cfg.Username,
 		Password: cfg.Password,
@@ -38,18 +39,18 @@ func Load(cfg config.MongoDB) (*Source, error) {
 	}
 
 	db := client.Database(cfg.Database)
-	return &Source{
+	return &SnapshotStore{
 		cfg: cfg,
 		db:  db,
 	}, nil
 }
 
-func (s *Source) Close() error {
+func (s *SnapshotStore) Close() error {
 	// MongoDB doesn't need to be closed.
 	return nil
 }
 
-func (s *Source) Run(ctx context.Context, writer writers.Writer) error {
+func (s *SnapshotStore) Run(ctx context.Context, writer writers.Writer) error {
 	for _, collection := range s.cfg.Collections {
 		snapshotStartTime := time.Now()