-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
d489e93
commit 74e72fb
Showing
5 changed files
with
277 additions
and
57 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| package looper | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "time" | ||
| ) | ||
|
|
||
| var ( | ||
| ErrFailedToConnectToLocker = errors.New("looper - failed to connect to locker") | ||
| ErrFailedToObtainLock = errors.New("looper - failed to obtain lock") | ||
| ErrFailedToReleaseLock = errors.New("looper - failed to release lock") | ||
| ErrFailedToCreateLockTable = errors.New("looper - failed to create lock table") | ||
| ErrFailedToCheckLockExistence = errors.New("looper - failed to check lock existence") | ||
| ) | ||
|
|
||
| type lockerKind int | ||
|
|
||
| const ( | ||
| lockerNop lockerKind = iota | ||
| lockerRedis | ||
| ) | ||
|
|
||
| // Lock if an error is returned by lock, the job will not be scheduled. | ||
| type locker interface { | ||
| lock(ctx context.Context, key string, timeout time.Duration) (lock, error) | ||
| } | ||
|
|
||
| // lock represents an obtained lock | ||
| type lock interface { | ||
| unlock(ctx context.Context) error | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| package looper | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
| ) | ||
|
|
||
| func newNopLocker() locker { | ||
| return &nopLocker{} | ||
| } | ||
|
|
||
| // Locker | ||
| var _ locker = (*nopLocker)(nil) | ||
|
|
||
| type nopLocker struct{} | ||
|
|
||
| func (r *nopLocker) lock(ctx context.Context, key string, timeout time.Duration) (lock, error) { | ||
| return &nopLock{}, nil | ||
| } | ||
|
|
||
| // Lock | ||
| var _ lock = (*nopLock)(nil) | ||
|
|
||
| type nopLock struct{} | ||
|
|
||
| func (r *nopLock) unlock(ctx context.Context) error { | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,161 @@ | ||
| package looper | ||
|
|
||
| import ( | ||
| "context" | ||
| "database/sql" | ||
| "fmt" | ||
| "time" | ||
| ) | ||
|
|
||
| const defaultTableName = "looper_lock" | ||
|
|
||
| // PostgresLocker provides an implementation of the Locker interface using | ||
| // a PostgreSQL table for storage. | ||
| func PostgresLocker(ctx context.Context, db *sql.DB, table string) (locker, error) { | ||
| err := db.PingContext(ctx) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("%w: %v", ErrFailedToConnectToLocker, err) | ||
| } | ||
|
|
||
| if table == "" { | ||
| table = defaultTableName | ||
| } | ||
| // Ensure the lock table exists, create it if necessary | ||
| err = createLockTable(ctx, db, table) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| pl := &postgresLocker{ | ||
| db: db, | ||
| table: table, | ||
| } | ||
|
|
||
| return pl, nil | ||
| } | ||
|
|
||
| // Locker | ||
| var _ locker = (*postgresLocker)(nil) | ||
|
|
||
| type postgresLocker struct { | ||
| db *sql.DB | ||
| table string | ||
| } | ||
|
|
||
| func createLockTable(ctx context.Context, db *sql.DB, table string) error { | ||
| var tableExists bool | ||
| err := db.QueryRowContext( | ||
| ctx, | ||
| fmt.Sprintf(` | ||
| SELECT EXISTS ( | ||
| SELECT 1 | ||
| FROM information_schema.tables | ||
| WHERE table_name = '%s' | ||
| );`, | ||
| table, | ||
| ), | ||
| ).Scan(&tableExists) | ||
| if err != nil { | ||
| return fmt.Errorf("%w: %v", ErrFailedToCheckLockExistence, err) | ||
| } | ||
|
|
||
| if !tableExists { | ||
| _, err := db.ExecContext( | ||
| ctx, | ||
| fmt.Sprintf(` | ||
| CREATE TABLE %s ( | ||
| job_name VARCHAR(255) PRIMARY KEY, | ||
| created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT (now() AT TIME ZONE 'UTC') | ||
| );`, | ||
| table, | ||
| )) | ||
| if err != nil { | ||
| return fmt.Errorf("%w: %v", ErrFailedToCreateLockTable, err) | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (p *postgresLocker) lock( | ||
| ctx context.Context, | ||
| key string, | ||
| timeout time.Duration, | ||
| ) (lock, error) { | ||
| // Create a row in the lock table to acquire the lock | ||
| _, err := p.db.ExecContext( | ||
| ctx, | ||
| fmt.Sprintf(` | ||
| INSERT INTO %s (job_name) | ||
| VALUES ('%s');`, | ||
| p.table, | ||
| key, | ||
| )) | ||
| if err != nil { | ||
| var createdAt time.Time | ||
| err := p.db.QueryRowContext( | ||
| ctx, | ||
| fmt.Sprintf(` | ||
| SELECT created_at | ||
| FROM %s | ||
| WHERE job_name = '%s';`, | ||
| p.table, | ||
| key, | ||
| )).Scan(&createdAt) | ||
| if err != nil { | ||
| return nil, ErrFailedToCheckLockExistence | ||
| } | ||
|
|
||
| if createdAt.Before(time.Now().Add(-timeout)) { | ||
| _, err := p.db.ExecContext( | ||
| ctx, | ||
| fmt.Sprintf(` | ||
| DELETE FROM %s | ||
| WHERE job_name = '%s';`, | ||
| p.table, | ||
| key, | ||
| )) | ||
| if err != nil { | ||
| return nil, ErrFailedToReleaseLock | ||
| } | ||
|
|
||
| return p.lock(ctx, key, timeout) | ||
| } | ||
|
|
||
| return nil, ErrFailedToObtainLock | ||
| } | ||
|
|
||
| pl := &postgresLock{ | ||
| db: p.db, | ||
| table: p.table, | ||
| key: key, | ||
| } | ||
|
|
||
| return pl, nil | ||
| } | ||
|
|
||
| // Lock | ||
| var _ lock = (*postgresLock)(nil) | ||
|
|
||
| type postgresLock struct { | ||
| db *sql.DB | ||
| table string | ||
| key string | ||
| } | ||
|
|
||
| func (p *postgresLock) unlock(ctx context.Context) error { | ||
| // Release the lock by deleting the row | ||
| _, err := p.db.ExecContext( | ||
| ctx, | ||
| fmt.Sprintf(` | ||
| DELETE FROM %s | ||
| WHERE job_name = '%s';`, | ||
| p.table, | ||
| p.key, | ||
| )) | ||
| if err != nil { | ||
| return ErrFailedToReleaseLock | ||
| } | ||
|
|
||
| return nil | ||
| } |
Oops, something went wrong.