proper handling of maxBytesPerFile for reads

[mreiferson]

fixes up and replaces #15

[ploxiln]

I attempted to make a test that fails without this branch ... I can get the current master branch to log some warnings and errors, but it still works out, so I had the test check for *.bad files at the end:

    diskqueue_test.go:76: ERROR: DISKQUEUE(test_disk_queue_resize1609301793) reading at 112 of /tmp/nsq-test-1609301793668354309126785195/test_disk_queue_resize1609301793.diskqueue.000000.dat - EOF
    diskqueue_test.go:76: WARNING: DISKQUEUE(test_disk_queue_resize1609301793) jump to next file and saving bad file as /tmp/nsq-test-1609301793668354309126785195/test_disk_queue_resize1609301793.diskqueue.000000.dat.bad
    diskqueue_test.go:76: INFO: DISKQUEUE(test_disk_queue_resize1609301793): readOne() opened /tmp/nsq-test-1609301793668354309126785195/test_disk_queue_resize1609301793.diskqueue.000001.dat
    diskqueue_test.go:76: ERROR: DISKQUEUE(test_disk_queue_resize1609301793) reading at 140 of /tmp/nsq-test-1609301793668354309126785195/test_disk_queue_resize1609301793.diskqueue.000001.dat - EOF
    diskqueue_test.go:76: WARNING: DISKQUEUE(test_disk_queue_resize1609301793) jump to next file and saving bad file as /tmp/nsq-test-1609301793668354309126785195/test_disk_queue_resize1609301793.diskqueue.000001.dat.bad
    diskqueue_test.go:76: INFO: DISKQUEUE(test_disk_queue_resize1609301793): closing
    diskqueue_test.go:76: INFO: DISKQUEUE(test_disk_queue_resize1609301793): closing ... ioLoop
    diskqueue_test.go:23: diskqueue_test.go:184:

               []string{} (expected)

            != []string{"/tmp/nsq-test-1609301793668354309126785195/test_disk_queue_resize1609301793.diskqueue.000000.dat.bad", "/tmp/nsq-test-1609301793668354309126785195/test_disk_queue_resize1609301793.diskqueue.000001.dat.bad"} (actual)

and here's the test:

func TestDiskQueueResize(t *testing.T) {
    l := NewTestLogger(t)
    dqName := "test_disk_queue_resize" + strconv.Itoa(int(time.Now().Unix()))
    tmpDir, err := ioutil.TempDir("", fmt.Sprintf("nsq-test-%d", time.Now().UnixNano()))
    if err != nil {
        panic(err)
    }
    defer os.RemoveAll(tmpDir)
    msg := []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
    ml := int64(len(msg))
    dq := New(dqName, tmpDir, 8*(ml+4), int32(ml), 1<<10, 2500, time.Second, l)
    NotNil(t, dq)
    Equal(t, int64(0), dq.Depth())

    for i := 0; i < 8; i++ {
        msg[0] = byte(i)
        err := dq.Put(msg)
        Nil(t, err)
    }
    Equal(t, int64(1), dq.(*diskQueue).writeFileNum)
    Equal(t, int64(0), dq.(*diskQueue).writePos)
    Equal(t, int64(8), dq.Depth())

    dq.Close()
    dq = New(dqName, tmpDir, 10*(ml+4), int32(ml), 1<<10, 2500, time.Second, l)

    for i := 0; i < 10; i++ {
        msg[0] = byte(20 + i)
        err := dq.Put(msg)
        Nil(t, err)
    }
    Equal(t, int64(2), dq.(*diskQueue).writeFileNum)
    Equal(t, int64(0), dq.(*diskQueue).writePos)
    Equal(t, int64(18), dq.Depth())

    for i := 0; i < 8; i++ {
        msg[0] = byte(i)
        Equal(t, msg, <-dq.ReadChan())
    }
    for i := 0; i < 10; i++ {
        msg[0] = byte(20 + i)
        Equal(t, msg, <-dq.ReadChan())
    }
    Equal(t, int64(0), dq.Depth())
    dq.Close()

    // make sure there aren't "bad" files due to read logic errors
    files, err := filepath.Glob(filepath.Join(tmpDir, dqName + "*.bad"))
    Nil(t, err)
    // empty files slice is actually nil, length check is less confusing
    if len(files) > 0 {
        Equal(t, []string{}, files)
    }
}

[mreiferson]

thanks, added that test here