@@ -60,14 +60,13 @@ type objState struct {
6060// StorageImpl offers a common interface for object marshaling/unmarshaling operations and
6161// hides all the storage-related operations behind it.
6262type StorageImpl struct {
63- appFs afero.Fs
64- pool * sqlitemigration.Pool
65- locks utils.MapMutex [string ]
66- processor Processor
67- root string
68- scheme * runtime.Scheme
69- versioner storage.Versioner
70- watchDispatcher * WatchDispatcher
63+ appFs afero.Fs
64+ pool * sqlitemigration.Pool
65+ locks utils.MapMutex [string ]
66+ processor Processor
67+ root string
68+ scheme * runtime.Scheme
69+ versioner storage.Versioner
7170}
7271
7372// StorageQuerier wraps the storage.Interface and adds some extra methods which are used by the storage implementation.
@@ -82,23 +81,19 @@ var _ storage.Interface = &StorageImpl{}
8281
8382var _ StorageQuerier = & StorageImpl {}
8483
85- func NewStorageImpl (appFs afero.Fs , root string , pool * sqlitemigration.Pool , watchDispatcher * WatchDispatcher , scheme * runtime.Scheme ) StorageQuerier {
86- return NewStorageImplWithCollector (appFs , root , pool , watchDispatcher , scheme , DefaultProcessor {})
84+ func NewStorageImpl (appFs afero.Fs , root string , pool * sqlitemigration.Pool , scheme * runtime.Scheme ) StorageQuerier {
85+ return NewStorageImplWithCollector (appFs , root , pool , scheme , DefaultProcessor {})
8786}
8887
89- func NewStorageImplWithCollector (appFs afero.Fs , root string , conn * sqlitemigration.Pool , watchDispatcher * WatchDispatcher , scheme * runtime.Scheme , processor Processor ) StorageQuerier {
90- if watchDispatcher == nil {
91- watchDispatcher = NewWatchDispatcher ()
92- }
88+ func NewStorageImplWithCollector (appFs afero.Fs , root string , conn * sqlitemigration.Pool , scheme * runtime.Scheme , processor Processor ) StorageQuerier {
9389 storageImpl := & StorageImpl {
94- appFs : appFs ,
95- pool : conn ,
96- locks : utils .NewMapMutex [string ](),
97- processor : processor ,
98- root : root ,
99- scheme : scheme ,
100- versioner : storage.APIObjectVersioner {},
101- watchDispatcher : watchDispatcher ,
90+ appFs : appFs ,
91+ pool : conn ,
92+ locks : utils .NewMapMutex [string ](),
93+ processor : processor ,
94+ root : root ,
95+ scheme : scheme ,
96+ versioner : storage.APIObjectVersioner {},
10297 }
10398 processor .SetStorage (storageImpl )
10499 return storageImpl
@@ -260,8 +255,6 @@ func (s *StorageImpl) CreateWithConn(ctx context.Context, conn *sqlite.Conn, key
260255 if err := s .processor .AfterCreate (ctx , conn , obj ); err != nil {
261256 return fmt .Errorf ("processor.AfterCreate: %w" , err )
262257 }
263- // publish event to watchers
264- s .watchDispatcher .Added (key , metaOut , obj )
265258 return nil
266259}
267260
@@ -309,8 +302,6 @@ func (s *StorageImpl) delete(ctx context.Context, conn *sqlite.Conn, key string,
309302 if err := s .appFs .Remove (makePayloadPath (p )); err != nil {
310303 logger .L ().Ctx (ctx ).Error ("Delete - remove json file failed" , helpers .Error (err ), helpers .String ("key" , key ))
311304 }
312- // publish event to watchers
313- s .watchDispatcher .Deleted (key , metaOut )
314305 return nil
315306}
316307
@@ -321,20 +312,8 @@ func (s *StorageImpl) delete(ctx context.Context, conn *sqlite.Conn, key string,
321312// (e.g. reconnecting without missing any updates).
322313// If resource version is "0", this interface will get current object at given key
323314// and send it in an "ADDED" event, before watch starts.
324- func (s * StorageImpl ) Watch (ctx context.Context , key string , opts storage.ListOptions ) (watch.Interface , error ) {
325- _ , span := otel .Tracer ("" ).Start (ctx , "StorageImpl.Watch" )
326- span .SetAttributes (attribute .String ("key" , key ))
327- defer span .End ()
328- _ , _ , _ , namespace , _ := pathToKeys (key )
329- if namespace != "" {
330- // FIXME find an alternative to fix NS deletion
331- logger .L ().Debug ("rejecting Watch called with namespace" , helpers .String ("key" , key ), helpers .String ("namespace" , namespace ))
332- return watch .NewEmptyWatch (), nil
333- }
334- // TODO(ttimonen) Should we do ctx.WithoutCancel; or does the parent ctx lifetime match with expectations?
335- nw := newWatcher (ctx , opts .ResourceVersion == softwarecomposition .ResourceVersionFullSpec )
336- s .watchDispatcher .Register (key , nw )
337- return nw , nil
315+ func (s * StorageImpl ) Watch (_ context.Context , _ string , _ storage.ListOptions ) (watch.Interface , error ) {
316+ return nil , nil // watch disabled
338317}
339318
340319// Get unmarshals object found at key into objPtr. On a not found error, will either
@@ -776,8 +755,6 @@ func (s *StorageImpl) GuaranteedUpdateWithConn(
776755 logger .L ().Ctx (ctx ).Error ("GuaranteedUpdate - save object failed" , helpers .Error (err ), helpers .String ("key" , key ))
777756 return err
778757 }
779- // Only successful updates should produce modification events
780- s .watchDispatcher .Modified (key , metaOut , ret )
781758 return nil
782759 }
783760}
@@ -910,7 +887,7 @@ func (immutableStorage) Delete(_ context.Context, key string, _ runtime.Object,
910887
911888// Watch is not supported for immutable objects. Objects are generated on the fly and not stored.
912889func (immutableStorage ) Watch (_ context.Context , _ string , _ storage.ListOptions ) (watch.Interface , error ) {
913- return watch . NewEmptyWatch () , nil
890+ return nil , nil // watch disabled
914891}
915892
916893// GuaranteedUpdate is not supported for immutable objects. Objects are generated on the fly and not stored.
0 commit comments