3pp-mirror/pomerium
1
0
Fork 0
mirror of https://github.com/pomerium/pomerium.git synced 2025-08-06 02:09:15 +02:00
Code Issues Projects Releases Packages Wiki Activity

use origin context

Browse source
This commit is contained in:
Caleb Doxsey 2024-04-23 09:23:01 -06:00
parent 1a5a9f1758
commit 27b3a39f2b
2 changed files with 18 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

4
internal/identity/manager/manager.go
View file

@ -82,12 +82,12 @@ func (mgr *Manager) RunLeased(ctx context.Context) error {
}) })
eg, ctx := errgroup.WithContext(ctx) eg, ctx := errgroup.WithContext(ctx)
eg.Go(func() error { eg.Go(func() error {
sessionSyncer := newSessionSyncer(ctx, mgr) sessionSyncer := newSessionSyncer(mgr)
defer sessionSyncer.Close() defer sessionSyncer.Close()
return fmt.Errorf("session syncer error: %w", sessionSyncer.Run(ctx)) return fmt.Errorf("session syncer error: %w", sessionSyncer.Run(ctx))
}) })
eg.Go(func() error { eg.Go(func() error {
userSyncer := newUserSyncer(ctx, mgr) userSyncer := newUserSyncer(mgr)
defer userSyncer.Close() defer userSyncer.Close()
return fmt.Errorf("user syncer error: %w", userSyncer.Run(ctx)) return fmt.Errorf("user syncer error: %w", userSyncer.Run(ctx))
}) })

34
internal/identity/manager/sync.go
View file

@ -11,68 +11,66 @@ import (
) )
type sessionSyncerHandler struct { type sessionSyncerHandler struct {
ctx context.Context
mgr *Manager mgr *Manager
} }
func newSessionSyncer(ctx context.Context, mgr *Manager) *databroker.Syncer { func newSessionSyncer(mgr *Manager) *databroker.Syncer {
return databroker.NewSyncer("identity_manager/sessions", sessionSyncerHandler{ctx: ctx, mgr: mgr}, return databroker.NewSyncer("identity_manager/sessions", sessionSyncerHandler{mgr: mgr},
databroker.WithTypeURL(grpcutil.GetTypeURL(new(session.Session)))) databroker.WithTypeURL(grpcutil.GetTypeURL(new(session.Session))))
} }
func (h sessionSyncerHandler) ClearRecords(_ context.Context) { func (h sessionSyncerHandler) ClearRecords(ctx context.Context) {
h.mgr.onDeleteAllSessions(h.ctx) h.mgr.onDeleteAllSessions(ctx)
} }
func (h sessionSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient { func (h sessionSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {
return h.mgr.cfg.Load().dataBrokerClient return h.mgr.cfg.Load().dataBrokerClient
} }
func (h sessionSyncerHandler) UpdateRecords(_ context.Context, _ uint64, records []*databroker.Record) { func (h sessionSyncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []*databroker.Record) {
for _, record := range records { for _, record := range records {
if record.GetDeletedAt() != nil { if record.GetDeletedAt() != nil {
h.mgr.onDeleteSession(h.ctx, record.GetId()) h.mgr.onDeleteSession(ctx, record.GetId())
} else { } else {
var s session.Session var s session.Session
err := record.Data.UnmarshalTo(&s) err := record.Data.UnmarshalTo(&s)
if err != nil { if err != nil {
log.Ctx(h.ctx).Warn().Err(err).Msg("invalid data in session record, ignoring") log.Ctx(ctx).Warn().Err(err).Msg("invalid data in session record, ignoring")
} else { } else {
h.mgr.onUpdateSession(h.ctx, &s) h.mgr.onUpdateSession(ctx, &s)
} }
} }
} }
} }
type userSyncerHandler struct { type userSyncerHandler struct {
ctx context.Context
mgr *Manager mgr *Manager
} }
func newUserSyncer(ctx context.Context, mgr *Manager) *databroker.Syncer { func newUserSyncer(mgr *Manager) *databroker.Syncer {
return databroker.NewSyncer("identity_manager/users", userSyncerHandler{ctx: ctx, mgr: mgr}, return databroker.NewSyncer("identity_manager/users", userSyncerHandler{mgr: mgr},
databroker.WithTypeURL(grpcutil.GetTypeURL(new(user.User)))) databroker.WithTypeURL(grpcutil.GetTypeURL(new(user.User))))
} }
func (h userSyncerHandler) ClearRecords(_ context.Context) { func (h userSyncerHandler) ClearRecords(ctx context.Context) {
h.mgr.onDeleteAllUsers(h.ctx) h.mgr.onDeleteAllUsers(ctx)
} }
func (h userSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient { func (h userSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {
return h.mgr.cfg.Load().dataBrokerClient return h.mgr.cfg.Load().dataBrokerClient
} }
func (h userSyncerHandler) UpdateRecords(_ context.Context, _ uint64, records []*databroker.Record) { func (h userSyncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []*databroker.Record) {
for _, record := range records { for _, record := range records {
if record.GetDeletedAt() != nil { if record.GetDeletedAt() != nil {
h.mgr.onDeleteUser(h.ctx, record.GetId()) h.mgr.onDeleteUser(ctx, record.GetId())
} else { } else {
var u user.User var u user.User
err := record.Data.UnmarshalTo(&u) err := record.Data.UnmarshalTo(&u)
if err != nil { if err != nil {
log.Ctx(h.ctx).Warn().Err(err).Msg("invalid data in user record, ignoring") log.Ctx(ctx).Warn().Err(err).Msg("invalid data in user record, ignoring")
} else { } else {
h.mgr.onUpdateUser(h.ctx, &u) h.mgr.onUpdateUser(ctx, &u)
} }
} }
} }