StartDownstreamServer starts a new downstream server with the given reader and writer
(reader io.Reader, writer io.Writer, options *DownstreamOptions)
| 43 | |
| 44 | // StartDownstreamServer starts a new downstream server with the given reader and writer |
| 45 | func StartDownstreamServer(reader io.Reader, writer io.Writer, options *DownstreamOptions) error { |
| 46 | pipe := util.NewStdStreamJoint(reader, writer, options.ExitOnClose) |
| 47 | lis := util.NewStdinListener() |
| 48 | done := make(chan error) |
| 49 | |
| 50 | // Compile ignore paths |
| 51 | ignoreMatcher, err := ignoreparser.CompilePaths(options.ExcludePaths, logpkg.Discard) |
| 52 | if err != nil { |
| 53 | return errors.Wrap(err, "compile paths") |
| 54 | } |
| 55 | |
| 56 | go func() { |
| 57 | s := grpc.NewServer() |
| 58 | downStream := &Downstream{ |
| 59 | options: options, |
| 60 | ignoreMatcher: ignoreMatcher, |
| 61 | events: make(chan notify.EventInfo, 1000), |
| 62 | changes: map[string]bool{}, |
| 63 | ping: &pingtimeout.PingTimeout{}, |
| 64 | } |
| 65 | |
| 66 | if options.Ping { |
| 67 | doneChan := make(chan struct{}) |
| 68 | defer close(doneChan) |
| 69 | downStream.ping.Start(doneChan) |
| 70 | } |
| 71 | |
| 72 | remote.RegisterDownstreamServer(s, downStream) |
| 73 | reflection.Register(s) |
| 74 | |
| 75 | // start watcher if this we should use it |
| 76 | watchStop := make(chan struct{}) |
| 77 | if !options.Polling { |
| 78 | stderrlog.Infof("Use inotify as watching method in container") |
| 79 | |
| 80 | go func() { |
| 81 | // set up a watchpoint listening for events within a directory tree rooted at specified directory |
| 82 | watchPath := options.RemotePath + "/..." |
| 83 | if options.NoRecursiveWatch { |
| 84 | watchPath = options.RemotePath |
| 85 | } |
| 86 | |
| 87 | err := notify.WatchWithFilter(watchPath, downStream.events, func(s string) bool { |
| 88 | if ignoreMatcher == nil || ignoreMatcher.RequireFullScan() { |
| 89 | return false |
| 90 | } |
| 91 | |
| 92 | stat, err := os.Stat(s) |
| 93 | if err != nil { |
| 94 | return false |
| 95 | } |
| 96 | |
| 97 | return ignoreMatcher.Matches(s[len(options.RemotePath):], stat.IsDir()) |
| 98 | }, notify.All) |
| 99 | if err != nil { |
| 100 | log.Fatalf("error watching path %s: %v", options.RemotePath, err) |
| 101 | return |
| 102 | } |