| 44 | } |
| 45 | |
| 46 | func (m *messageDB) start(s sender) error { |
| 47 | var err error |
| 48 | path := fmt.Sprintf("%s/%s", m.root, util.MessageDBPath) |
| 49 | if m.stable, err = leveldb.OpenFile(path, nil); err != nil { |
| 50 | log.Errorf("db openFile:%v, err:%s", path, err.Error()) |
| 51 | return errors.Trace(err) |
| 52 | } |
| 53 | path = fmt.Sprintf("%s/%s", m.root, util.MessageLogDBPath) |
| 54 | if m.queue, err = leveldb.OpenFile(path, nil); err != nil { |
| 55 | log.Errorf("db openFile:%v, err:%s", path, err.Error()) |
| 56 | return errors.Trace(err) |
| 57 | } |
| 58 | |
| 59 | start := util.EncodeInt64(0) |
| 60 | end := util.EncodeInt64(math.MaxInt64) |
| 61 | |
| 62 | it := m.queue.NewIterator(&lu.Range{Start: start, Limit: end}, nil) |
| 63 | for it.Next(); it.Valid(); it.Next() { |
| 64 | m.addRetry(util.DecodeInt64(it.Key())) |
| 65 | } |
| 66 | |
| 67 | m.sender = s |
| 68 | |
| 69 | go m.repush() |
| 70 | |
| 71 | return nil |
| 72 | } |
| 73 | |
| 74 | // 定时推送之前失败过的消息 |
| 75 | func (m *messageDB) repush() { |