testFileWriterCompareInOut sends numRecords records and wait for the given duration between each send, to a FileWriter where PathString is set to the concatenation of the path components.
(numRecords int, wait, rotate time.Duration, comps ...string)
| 99 | // duration between each send, to a FileWriter where PathString is set to the |
| 100 | // concatenation of the path components. |
| 101 | func testFileWriterCompareInOut(numRecords int, wait, rotate time.Duration, comps ...string) func(*testing.T) { |
| 102 | return func(t *testing.T) { |
| 103 | tmpDir := t.TempDir() |
| 104 | cfg := baker.OutputParams{ |
| 105 | Fields: []baker.FieldIndex{1}, |
| 106 | ComponentParams: baker.ComponentParams{ |
| 107 | DecodedConfig: &output.FileWriterConfig{ |
| 108 | PathString: filepath.Join(append([]string{tmpDir}, comps...)...), |
| 109 | RotateInterval: rotate, |
| 110 | }, |
| 111 | }, |
| 112 | } |
| 113 | fw, err := output.NewFileWriter(cfg) |
| 114 | if err != nil { |
| 115 | t.Fatal(err) |
| 116 | } |
| 117 | |
| 118 | // Send records to inch and keep track of them in a map for later comparison. |
| 119 | inch := make(chan baker.OutputRecord) |
| 120 | sentRecords := make(map[int]string) |
| 121 | go func() { |
| 122 | for i := 0; i < numRecords; i++ { |
| 123 | record := fmt.Sprintf("foo,%d,bar", i) |
| 124 | inch <- baker.OutputRecord{Fields: []string{strconv.Itoa(i)}, Record: []byte(record)} |
| 125 | sentRecords[i] = record |
| 126 | time.Sleep(wait) |
| 127 | } |
| 128 | close(inch) |
| 129 | }() |
| 130 | |
| 131 | upch := make(chan string) |
| 132 | errc := make(chan error, 1) |
| 133 | go func() { |
| 134 | errc <- fw.Run(inch, upch) |
| 135 | close(upch) |
| 136 | }() |
| 137 | |
| 138 | // Drain the channel containing the uploaded paths. |
| 139 | uploaded := make(map[string]struct{}) |
| 140 | for p := range upch { |
| 141 | if _, ok := uploaded[p]; ok { |
| 142 | t.Errorf("file uploaded twice: %q", p) |
| 143 | } |
| 144 | uploaded[p] = struct{}{} |
| 145 | } |
| 146 | |
| 147 | if err := <-errc; err != nil { |
| 148 | t.Fatalf("fw.Run() error: %v", err) |
| 149 | } |
| 150 | |
| 151 | // Verify that the set of records sent to the output is equal to the set of |
| 152 | // records present in the file(s) sent to the uploader. |
| 153 | uploadedRecords := make(map[int]string) |
| 154 | |
| 155 | for p := range uploaded { |
| 156 | f, err := os.Open(p) |
| 157 | if err != nil { |
| 158 | t.Fatalf("can't open uploaded path: %s", err) |
no test coverage detected