| 213 | } |
| 214 | |
| 215 | func TestGoingHam(t *testing.T) { |
| 216 | prePoolOpts := goleak.IgnoreCurrent() |
| 217 | |
| 218 | p := NewPool(&Config{ |
| 219 | MaxWorkers: 1000, |
| 220 | QueueDepth: 10000, |
| 221 | }) |
| 222 | opts := goleak.IgnoreCurrent() |
| 223 | |
| 224 | wg := &sync.WaitGroup{} |
| 225 | |
| 226 | for i := 0; i < 1000; i++ { |
| 227 | wg.Add(1) |
| 228 | go func() { |
| 229 | ret := []byte{0x01, 0x03, 0x04} |
| 230 | fn := func(_ context.Context, payload interface{}) (interface{}, error) { |
| 231 | i := payload.(int) |
| 232 | |
| 233 | time.Sleep(time.Duration(rand.Uint32()%100) * time.Millisecond) |
| 234 | if i == 5 { |
| 235 | return ret, nil |
| 236 | } |
| 237 | return nil, nil |
| 238 | } |
| 239 | payloads := []interface{}{1, 2, 3, 4, 5} |
| 240 | |
| 241 | msg, funcErrs, err := p.RunJobs(context.Background(), payloads, fn) |
| 242 | assert.NoError(t, err) |
| 243 | assert.Nil(t, funcErrs) |
| 244 | require.Len(t, msg, 1) |
| 245 | assert.Equal(t, ret, msg[0]) |
| 246 | wg.Done() |
| 247 | }() |
| 248 | } |
| 249 | |
| 250 | wg.Wait() |
| 251 | goleak.VerifyNone(t, opts) |
| 252 | |
| 253 | p.Shutdown() |
| 254 | goleak.VerifyNone(t, prePoolOpts) |
| 255 | } |
| 256 | |
| 257 | func TestOverloadingASmallPool(t *testing.T) { |
| 258 | prePoolOpts := goleak.IgnoreCurrent() |