setupWatcher starts watching events for sync
(ctx context.Context, source model.PathSyncSource, target model.PathSyncTarget)
| 78 | |
| 79 | // setupWatcher starts watching events for sync |
| 80 | func (s *Sync) setupWatcher(ctx context.Context, source model.PathSyncSource, target model.PathSyncTarget) (chan bool, error) { |
| 81 | |
| 82 | var err error |
| 83 | watchObject, err := source.Watch(ctx, "") |
| 84 | if err != nil { |
| 85 | log.Logger(ctx).Error("Error While Setting up Watcher on source", zap.Any("source", source), zap.Error(err)) |
| 86 | return nil, err |
| 87 | } |
| 88 | |
| 89 | //s.doneChans = append(s.doneChans, watchObject.DoneChan) |
| 90 | |
| 91 | var inputClosed bool |
| 92 | input := make(chan model.EventInfo) |
| 93 | inputCloser := make(chan bool) |
| 94 | |
| 95 | out := input |
| 96 | // If EchoFilter is registered, pipe |
| 97 | if s.echoFilter != nil { |
| 98 | out = s.echoFilter.Pipe(out) |
| 99 | } |
| 100 | |
| 101 | // If there are selective roots, pipe |
| 102 | if filters.NeedsSelectiveRootsFilter(s.Roots) { |
| 103 | out = filters.NewSelectiveRootsFilter(s.Roots).Pipe(out) |
| 104 | } |
| 105 | |
| 106 | // Finally Batch filtered events and register batcher for force-close session broadcast |
| 107 | batcher := filters.NewEventsBatcher(ctx, source, target, s.Ignores) |
| 108 | batcher.SetupChannels(s.statuses, s.runDone, s.cmd) |
| 109 | if s.watchConn != nil { |
| 110 | batcher.SetEndpointStatusChan(s.watchConn) |
| 111 | } |
| 112 | batcher.Batch(out, s.patchChan) |
| 113 | s.eventsBatchers = append(s.eventsBatchers, batcher) |
| 114 | |
| 115 | go func() { |
| 116 | // Wait for all events. |
| 117 | for { |
| 118 | select { |
| 119 | case event, ok := <-watchObject.Events(): |
| 120 | if !ok { |
| 121 | <-time.After(1 * time.Second) |
| 122 | continue |
| 123 | } |
| 124 | if !inputClosed { |
| 125 | input <- event |
| 126 | } |
| 127 | case err, ok := <-watchObject.Errors(): |
| 128 | if !ok { |
| 129 | <-time.After(5 * time.Second) |
| 130 | continue |
| 131 | } |
| 132 | if err != nil { |
| 133 | log.Logger(ctx).Error("Received error from watcher", zap.Error(err)) |
| 134 | if err.Error() == "API Not Supported" { |
| 135 | // Special case for S3 Client => Send Disconnected Status if API does not support events watch |
| 136 | log.Logger(ctx).Error("Endpoint does not support watching with this config", zap.Error(err)) |
| 137 | s.watchConn <- &model.EndpointStatus{ |
no test coverage detected