fix foreach

zhangzengfei

2021-11-12 d7485b186f72742ee0a1bf34e7fc818b53f20870

fix foreach

 api.go

@@ -52,7 +52,7 @@
}

// 遍历数据, 输出原始数据
func (ls *LogStore) ForEach(f func(v []byte) error) {
func (ls *LogStore) ForEach(f func(v *LogCon) error) {
    ls.store.ForEach(f)
}


 bolt_store.go

@@ -3,7 +3,6 @@
import (
    "errors"
    "github.com/boltdb/bolt"
    "strconv"
    "time"
)

@@ -243,7 +242,7 @@
    return tx.Commit()
}

func (b *BoltStore) ForEach(f func(v []byte) error) {
func (b *BoltStore) ForEach(f func(lc *LogCon) error) {
    tx, err := b.conn.Begin(true)
    if err != nil {
        return
@@ -254,7 +253,13 @@
    end, _ := b.LastIndex(confBucket)

    for ; start <= end; start++ {
        bucketName := bucketPre + strconv.Itoa(int(start))
        cLog := &confLog{}
        err := b.GetConfLog(start, cLog)
        if err != nil {
            return
        }

        bucketName := cLog.BucketName
        bucket := tx.Bucket([]byte(bucketName))
        if bucket == nil {
            return
@@ -263,7 +268,10 @@
        bucket.ForEach(func(k, v []byte) error {
            log := &Log{}
            err := decodeMsgPack(v, log)
            f(log.Data)
            if err == nil {
                f(&LogCon{cLog, log})
            }

            return err
        })
    }

 store_test.go

@@ -1,7 +1,6 @@
package boltcache

import (
    "strconv"
    "testing"
    "time"
)
@@ -18,19 +17,27 @@

    //go consume(ls)
    //
    for i := 0; i < 100; i++ {
        ls.ApplyLog([]byte("hello world " + strconv.Itoa(i)))
        time.Sleep(time.Millisecond * 100)
    }
    //for i := 0; i < 10; i++ {
    //    ls.ApplyLog([]byte("hello world " + strconv.Itoa(i)))
    //    time.Sleep(time.Millisecond * 100)
    //}

    ls.printLog("Size = ", ls.Size())

    ls.ForEach(func(v []byte) error {
        ls.printLog("val=:", string(v))
    var logs []*LogCon
    ls.ForEach(func(log *LogCon) error {
        ls.printLog("val=:", string(log.GetData()))

        logs = append(logs, log)
        time.Sleep(time.Millisecond * 100)

        return nil
    })

    consume(ls)
    for _, log := range logs {
        ls.Delete(log)
    }
    //consume(ls)
}

func consume(ls *LogStore) {