diff --git a/g/os/gfcache/gfcache.go b/g/os/gfcache/gfcache.go index 2e4f4eae7..26831d481 100644 --- a/g/os/gfcache/gfcache.go +++ b/g/os/gfcache/gfcache.go @@ -8,16 +8,14 @@ package gfcache import ( - "gitee.com/johng/gf/g/os/gcache" "gitee.com/johng/gf/g/container/gtype" - "gitee.com/johng/gf/g/os/gfsnotify" + "gitee.com/johng/gf/g/os/gcache" ) type Cache struct { cap *gtype.Int // 缓存容量(byte)，设置为0表示不限制 size *gtype.Int // 缓存大小(Byte) cache *gcache.Cache // 缓存对象 - notify *gfsnotify.Watcher // 文件监控管理对象 } const ( @@ -35,12 +33,10 @@ func New(cap ... int) *Cache { if len(cap) > 0 { c = cap[0] } - notify, _ := gfsnotify.New() return &Cache { cap : gtype.NewInt(c), size : gtype.NewInt(), cache : gcache.New(), - notify : notify, } } diff --git a/g/os/gfcache/gfcache_cache.go b/g/os/gfcache/gfcache_cache.go index c59fcc307..91981fa01 100644 --- a/g/os/gfcache/gfcache_cache.go +++ b/g/os/gfcache/gfcache_cache.go @@ -53,7 +53,7 @@ func (c *Cache) addMonitor(path string) { if c.cache.Get(path) != nil { return } - c.notify.Add(path, func(event *gfsnotify.Event) { + gfsnotify.Add(path, func(event *gfsnotify.Event) { //glog.Debug("gfcache:", event) r := c.cache.Get(path).([]byte) // 是否删除 diff --git a/g/os/gfsnotify/gfsnotify.go b/g/os/gfsnotify/gfsnotify.go index 2d53c7d55..bb8bdac90 100644 --- a/g/os/gfsnotify/gfsnotify.go +++ b/g/os/gfsnotify/gfsnotify.go @@ -5,10 +5,13 @@ // You can obtain one at https://gitee.com/johng/gf. // 文件监控. -// 使用时需要注意的是，一旦一个文件被删除，那么对其的监控将会失效；如果删除的是目录，那么该目录及其下的文件都将被递归删除监控。 package gfsnotify import ( + "container/list" + "errors" + "fmt" + "gitee.com/johng/gf/g/container/glist" "gitee.com/johng/gf/g/container/gmap" "gitee.com/johng/gf/g/container/gqueue" "gitee.com/johng/gf/g/container/gtype" @@ -22,19 +25,29 @@ import ( // 监听管理对象 type Watcher struct { - watcher *fsnotify.Watcher // 底层fsnotify对象 - events *gqueue.Queue // 过滤后的事件通知，不会出现重复事件 - closeChan chan struct{} // 关闭事件 - callbacks *gmap.StringInterfaceMap // 监听的回调函数 - cache *gcache.Cache // 缓存对象，用于事件重复过滤 + watcher *fsnotify.Watcher // 底层fsnotify对象 + events *gqueue.Queue // 过滤后的事件通知，不会出现重复事件 + closeChan chan struct{} // 关闭事件 + callbacks *gmap.StringInterfaceMap // 监听的回调函数 + cache *gcache.Cache // 缓存对象，用于事件重复过滤 +} + +// 注册的监听回调方法 +type Callback struct { + Id int // 唯一ID + Func func(event *Event) // 回调方法 + Path string // 监听的文件/目录 + elem *list.Element // 指向监听链表中的元素项位置 + parent *Callback // 父级callback，有这个属性表示该callback为被自动管理的callback + subs *glist.List // 子级回调对象指针列表 } // 监听事件对象 type Event struct { - event fsnotify.Event // 底层事件对象 - Path string // 文件绝对路径 - Op Op // 触发监听的文件操作 - Watcher *Watcher // 事件对应的监听对象 + event fsnotify.Event // 底层事件对象 + Path string // 文件绝对路径 + Op Op // 触发监听的文件操作 + Watcher *Watcher // 事件对应的监听对象 } // 按位进行识别的操作集合 @@ -51,21 +64,25 @@ const ( const ( REPEAT_EVENT_FILTER_INTERVAL = 1 // (毫秒)重复事件过滤间隔 - DEFAULT_WATCHER_COUNT = 8 // 默认创建的监控对象数量(使用哈希取模) + DEFAULT_WATCHER_COUNT = 2 // 默认创建的监控对象数量(使用哈希取模) ) var ( // 全局监听对象，方便应用端调用 - watchers []*Watcher + watchers []*Watcher + // 全局默认的监听watcher数量 + watcherCount int // 默认的watchers是否初始化，使用时才创建 - watcherInited = gtype.NewBool() + watcherInited = gtype.NewBool() + // 回调方法ID与对象指针的映射哈希表，用于根据ID快速查找回调对象 + callbackIdMap = gmap.NewIntInterfaceMap() ) // 初始化创建watcher对象，用于包默认管理监听 func initWatcher() { if !watcherInited.Set(true) { // 默认的创建的inotify数量 - watcherCount := gconv.Int(genv.Get("GF_INOTIFY_COUNT")) + watcherCount = gconv.Int(genv.Get("GF_INOTIFY_COUNT")) if watcherCount == 0 { watcherCount = gconv.Int(gcmd.Option.Get("gf.inotify-count")) } @@ -87,11 +104,11 @@ func initWatcher() { func New() (*Watcher, error) { if watch, err := fsnotify.NewWatcher(); err == nil { w := &Watcher { - cache : gcache.New(), - watcher : watch, - events : gqueue.New(), - closeChan : make(chan struct{}), - callbacks : gmap.NewStringInterfaceMap(), + cache : gcache.New(), + watcher : watch, + events : gqueue.New(), + closeChan : make(chan struct{}), + callbacks : gmap.NewStringInterfaceMap(), } w.startWatchLoop() w.startEventLoop() @@ -102,17 +119,29 @@ func New() (*Watcher, error) { } // 添加对指定文件/目录的监听，并给定回调函数；如果给定的是一个目录，默认递归监控。 -func Add(path string, callback func(event *Event), recursive...bool) error { - return getWatcherByPath(path).Add(path, callback, recursive...) +func Add(path string, callbackFunc func(event *Event), recursive...bool) (callback *Callback, err error) { + return getWatcherByPath(path).Add(path, callbackFunc, recursive...) } -// 移除监听，默认递归删除。 +// 递归移除对指定文件/目录的所有监听回调 func Remove(path string) error { return getWatcherByPath(path).Remove(path) } +// 根据指定的回调函数ID，移出指定的inotify回调函数 +func RemoveCallback(callbackId int) error { + callback := (*Callback)(nil) + if r := callbackIdMap.Get(callbackId); r != nil { + callback = r.(*Callback) + } + if callback == nil { + return errors.New(fmt.Sprintf(`callback for id %d not found`, callbackId)) + } + return getWatcherByPath(callback.Path).RemoveCallback(callbackId) +} + // 根据path计算对应的watcher对象 func getWatcherByPath(path string) *Watcher { initWatcher() - return watchers[ghash.BKDRHash([]byte(path)) % DEFAULT_WATCHER_COUNT] + return watchers[ghash.BKDRHash([]byte(path)) % uint32(watcherCount)] } diff --git a/g/os/gfsnotify/gfsnotify_watcher.go b/g/os/gfsnotify/gfsnotify_watcher.go index 9cb15f447..5d5ce4b13 100644 --- a/g/os/gfsnotify/gfsnotify_watcher.go +++ b/g/os/gfsnotify/gfsnotify_watcher.go @@ -10,6 +10,7 @@ import ( "errors" "fmt" "gitee.com/johng/gf/g/container/glist" + "gitee.com/johng/gf/g/os/gtime" ) // 关闭监听管理对象 @@ -20,13 +21,33 @@ func (w *Watcher) Close() { } // 添加对指定文件/目录的监听，并给定回调函数 -func (w *Watcher) addWatch(path string, callback func(event *Event)) error { +func (w *Watcher) addWatch(path string, calbackFunc func(event *Event), parentCallback *Callback) (callback *Callback, err error) { // 这里统一转换为当前系统的绝对路径，便于统一监控文件名称 t := fileRealPath(path) if t == "" { - return errors.New(fmt.Sprintf(`"%s" does not exist`, path)) + return nil, errors.New(fmt.Sprintf(`"%s" does not exist`, path)) } path = t + // 添加成功后会注册该callback id到全局的哈希表，并绑定到父级的注册回调中 + defer func() { + if err == nil { + if parentCallback == nil { + // 只有主callback才记录到id map中，因为子callback是自动管理的无需添加到全局id映射map中 + callbackIdMap.Set(callback.Id, callback) + } + if parentCallback != nil { + // 添加到直属父级的subs属性中，建立关联关系，便于后续删除 + parentCallback.subs.PushBack(callback) + } + } + }() + callback = &Callback { + Id : int(gtime.Nanosecond()), + Func : calbackFunc, + Path : path, + subs : glist.New(), + parent : parentCallback, + } // 注册回调函数 w.callbacks.LockFunc(func(m map[string]interface{}) { var result interface{} @@ -36,54 +57,109 @@ func (w *Watcher) addWatch(path string, callback func(event *Event)) error { } else { result = v } - result.(*glist.List).PushBack(callback) + callback.elem = result.(*glist.List).PushBack(callback) }) // 添加底层监听 w.watcher.Add(path) - return nil + return } -// 添加监控，path参数支持文件或者目录路径，recursive为非必需参数，默认为递归添加监控(当path为目录时) -func (w *Watcher) Add(path string, callback func(event *Event), recursive...bool) error { +// 添加监控，path参数支持文件或者目录路径，recursive为非必需参数，默认为递归添加监控(当path为目录时)。 +// 如果添加目录，这里只会返回目录的callback，按照callback删除时会递归删除。 +func (w *Watcher) addWithCallback(parentCallback *Callback, path string, callbackFunc func(event *Event), recursive...bool) (callback *Callback, err error) { + // 首先添加这个目录 + if callback, err = w.addWatch(path, callbackFunc, parentCallback); err != nil { + return nil, err + } + // 其次递归添加其下的文件/目录 if fileIsDir(path) && (len(recursive) == 0 || recursive[0]) { paths, _ := fileScanDir(path, "*", true) - list := []string{path} - list = append(list, paths...) - for _, v := range list { - if err := w.addWatch(v, callback); err != nil { - return err - } + for _, v := range paths { + w.addWatch(v, callbackFunc, callback) } - return nil - } else { - return w.addWatch(path, callback) } + return } - -// 移除监听 -func (w *Watcher) removeWatch(path string) error { - w.callbacks.Remove(path) - return w.watcher.Remove(path) +// 添加监控，path参数支持文件或者目录路径，recursive为非必需参数，默认为递归添加监控(当path为目录时)。 +// 如果添加目录，这里只会返回目录的callback，按照callback删除时会递归删除。 +func (w *Watcher) Add(path string, callbackFunc func(event *Event), recursive...bool) (callback *Callback, err error) { + return w.addWithCallback(nil, path, callbackFunc, recursive...) } -// 递归移除监听 +// 递归移除对指定文件/目录的所有监听回调 func (w *Watcher) Remove(path string) error { if fileIsDir(path) { paths, _ := fileScanDir(path, "*", true) - list := []string{path} - list = append(list, paths...) - for _, v := range list { - if err := w.removeWatch(v); err != nil { + paths = append(paths, path) + for _, v := range paths { + if err := w.removeAll(v); err != nil { return err } } return nil } else { - return w.removeWatch(path) + return w.removeAll(path) } } +// 移除对指定文件/目录的所有监听 +func (w *Watcher) removeAll(path string) error { + // 首先移除所有该path的回调注册 + if r := w.callbacks.Get(path); r != nil { + list := r.(*glist.List) + for { + if r := list.PopFront(); r != nil { + w.removeCallback(r.(*Callback)) + } else { + break + } + } + } + // 其次移除该path的监听注册 + w.callbacks.Remove(path) + // 最后移除底层的监听 + return w.watcher.Remove(path) +} + +// 根据指定的回调函数ID，移出指定的inotify回调函数 +func (w *Watcher) RemoveCallback(callbackId int) error { + callback := (*Callback)(nil) + if r := callbackIdMap.Get(callbackId); r != nil { + callback = r.(*Callback) + } + if callback == nil { + return errors.New(fmt.Sprintf(`callback for id %d not found`, callbackId)) + } + w.removeCallback(callback) + return nil +} + +// 移除对指定文件/目录的所有监听 +func (w *Watcher) removeCallback(callback *Callback) error { + if r := w.callbacks.Get(callback.Path); r != nil { + list := r.(*glist.List) + list.Remove(callback.elem) + // 如果存在子级callback，那么也一并递归删除 + if callback.subs.Len() > 0 { + for { + if r := callback.subs.PopFront(); r != nil { + w.removeCallback(r.(*Callback)) + } else { + break + } + } + } + // 如果该文件/目录的所有回调都被删除，那么移除监听 + if list.Len() == 0 { + return w.watcher.Remove(callback.Path) + } + } else { + return errors.New(fmt.Sprintf(`callbacks not found for "%s"`, callback.Path)) + } + return nil +} + // 监听循环 func (w *Watcher) startWatchLoop() { go func() { @@ -146,14 +222,16 @@ func (w *Watcher) startEventLoop() { callbacks := w.getCallbacks(event.Path) // 如果创建了新的目录，那么将这个目录递归添加到监控中 if event.IsCreate() && fileIsDir(event.Path) { - for _, callback := range callbacks.FrontAll() { - w.Add(event.Path, callback.(func(event *Event))) + for _, v := range callbacks.FrontAll() { + callback := v.(*Callback) + w.addWithCallback(callback, event.Path, callback.Func) } } + // 执行回调处理，异步处理 if callbacks != nil { go func(callbacks *glist.List) { - for _, callback := range callbacks.FrontAll() { - callback.(func(event *Event))(event) + for _, v := range callbacks.FrontAll() { + go v.(*Callback).Func(event) } }(callbacks) } diff --git a/g/os/gtime/gtime.go b/g/os/gtime/gtime.go index 45f8d95d7..3263b53da 100644 --- a/g/os/gtime/gtime.go +++ b/g/os/gtime/gtime.go @@ -34,6 +34,7 @@ const ( // 01/Nov/2018 11:50:28 // 01/Nov/2018:11:50:28 // 01/Nov/18 11:50:28 + // 01/Nov/18 11:50:28 TIME_REAGEX_PATTERN2 = `(\d{1,2}[-/][A-Za-z]{3,}[-/]\d{2,4})[:\sT-]*(\d{0,2}:{0,1}\d{0,2}:{0,1}\d{0,2}){0,1}\.{0,1}(\d{0,9})([\sZ]{0,1})([\+-]{0,1})([:\d]*)` ) @@ -130,72 +131,59 @@ func Datetime() string { return time.Now().Format("2006-01-02 15:04:05") } +// 解析日期字符串(支持'-'或'/'连接符号) +func parseDateStr(s string) (year, month, day int) { + array := strings.Split(s, "-") + if len(array) < 3 { + array = strings.Split(s, "/") + } + if len(array) >= 3 { + // 年是否为缩写，如果是，那么需要补上前缀 + year, _ = strconv.Atoi(array[0]) + if year < 100 { + year = int(time.Now().Year()/100)*100 + year + } + month, _ = strconv.Atoi(array[1]) + day, _ = strconv.Atoi(array[2]) + } + return +} + // 字符串转换为时间对象，第二个参数指定格式的format(如: Y-m-d H:i:s)，当指定第二个参数时同StrToTimeFormat方法 func StrToTime(str string, format...string) (*Time, error) { if len(format) > 0 { return StrToTimeFormat(str, format[0]) } - var year, month, day, hour, min, sec, nsec int + var year, month, day int + var hour, min, sec, nsec int var array, match []string var local = time.Local - match = timeRegex1.FindStringSubmatch(str) - if len(match) > 0 { + if match = timeRegex1.FindStringSubmatch(str); len(match) > 0 { for k, v := range match { match[k] = strings.TrimSpace(v) } - // 日期(支持'-'或'/'连接符号) - array = strings.Split(match[1], "-") - if len(array) < 3 { - array = strings.Split(match[1], "/") - } - if len(array) >= 3 { - // 年是否为缩写，如果是，那么需要补上前缀 - year, _ = strconv.Atoi(array[0]) - if year < 100 { - year = int(time.Now().Year()/100)*100 + year - } - month, _ = strconv.Atoi(array[1]) - day, _ = strconv.Atoi(array[2]) + year, month, day = parseDateStr(match[1]) + } else if match = timeRegex2.FindStringSubmatch(str); len(match) > 0 { + for k, v := range match { + match[k] = strings.TrimSpace(v) } + year, month, day = parseDateStr(match[1]) } else { - match = timeRegex2.FindStringSubmatch(str) - if len(match) > 0 { - for k, v := range match { - match[k] = strings.TrimSpace(v) - } - // 日期(支持'-'或'/'连接符号) - array = strings.Split(match[1], "-") - if len(array) < 3 { - array = strings.Split(match[1], "/") - } - if len(array) >= 3 { - day, _ = strconv.Atoi(array[0]) - if v, ok := monthMap[strings.ToLower(array[1])]; ok { - month = v - } else { - return nil, errors.New("invalid month:" + array[1]) - } - // 年是否为缩写，如果是，那么需要补上前缀 - year, _ = strconv.Atoi(array[2]) - if year < 100 { - year = int(time.Now().Year()/100)*100 + year - } - } - } - } - if len(match) == 0 { return nil, errors.New("unsupported time format") } // 时间 if len(match[2]) > 0 { - array = strings.Split(match[2], ":") - hour, _ = strconv.Atoi(array[0]) + s := strings.Replace(match[2], ":", "", -1) + if len(s) < 6 { + s += strings.Repeat("0", 6 - len(s)) + } + hour, _ = strconv.Atoi(s[0 : 2]) if len(array) >= 2 { - min, _ = strconv.Atoi(array[1]) + min, _ = strconv.Atoi(s[2 : 4]) } if len(array) >= 3 { - sec, _ = strconv.Atoi(array[2]) + sec, _ = strconv.Atoi(s[4 : 6]) } } // 纳秒，检查并执行位补齐 @@ -291,7 +279,8 @@ func StrToTimeLayout(str string, layout string) (*Time, error) { } } -// 从文本内容中解析时间，并返回解析成功的时间对象。注意当文本中存在多个时间时，会解析第一个。 +// 从字符串内容中(也可以是文件名称等等)解析时间，并返回解析成功的时间对象，否则返回nil。 +// 注意当内容中存在多个时间时，会解析第一个。 // format参数可以指定需要解析的时间格式。 func ParseTimeFromContent(content string, format...string) *Time { if len(format) > 0 { @@ -300,9 +289,9 @@ func ParseTimeFromContent(content string, format...string) *Time { } } else { if match := timeRegex1.FindStringSubmatch(content); len(match) >= 1 { - return NewFromStr(match[0]) + return NewFromStr(strings.Trim(match[0], "./_- \n\r")) } else if match := timeRegex2.FindStringSubmatch(content); len(match) >= 1 { - return NewFromStr(match[0]) + return NewFromStr(strings.Trim(match[0], "./_- \n\r")) } } return nil diff --git a/g/util/grand/grand.go b/g/util/grand/grand.go index a9d581fdf..ff1c66972 100644 --- a/g/util/grand/grand.go +++ b/g/util/grand/grand.go @@ -7,8 +7,10 @@ // 随机数管理 package grand -var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") -var digits = []rune("0123456789") +var ( + letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") + digits = []rune("0123456789") +) // 获得一个 min, max 之间的随机数(min <= x <= max) func Rand (min, max int) int { diff --git a/geg/os/gfsnotify/gfsnotify.go b/geg/os/gfsnotify/gfsnotify.go index 4683b9d70..9f378b597 100644 --- a/geg/os/gfsnotify/gfsnotify.go +++ b/geg/os/gfsnotify/gfsnotify.go @@ -6,7 +6,9 @@ import ( ) func main() { - err := gfsnotify.Add("/home/john/temp", func(event *gfsnotify.Event) { + // /home/john/temp 是一个目录，当然也可以指定文件 + path := "/home/john/temp" + _, err := gfsnotify.Add(path, func(event *gfsnotify.Event) { if event.IsCreate() { glog.Println("创建文件 : ", event.Path) } @@ -24,6 +26,8 @@ func main() { } glog.Println(event) }) + // 移除对该path的监听 + gfsnotify.Remove(path) if err != nil { glog.Fatalln(err) } else { diff --git a/geg/os/gfsnotify/gfsnotify_callback.go b/geg/os/gfsnotify/gfsnotify_callback.go new file mode 100644 index 000000000..ccb032d36 --- /dev/null +++ b/geg/os/gfsnotify/gfsnotify_callback.go @@ -0,0 +1,36 @@ +package main + +import ( + "gitee.com/johng/gf/g/os/gfsnotify" + "gitee.com/johng/gf/g/os/glog" + "gitee.com/johng/gf/g/os/gtime" + "time" +) + +func main() { + c1, err := gfsnotify.Add("/home/john/temp/log", func(event *gfsnotify.Event) { + glog.Println("callback1") + }) + if err != nil { + panic(err) + } + c2, err := gfsnotify.Add("/home/john/temp/log", func(event *gfsnotify.Event) { + glog.Println("callback2") + }) + if err != nil { + panic(err) + } + // 5秒后移除c1的回调函数注册，仅剩c2 + gtime.SetTimeout(5*time.Second, func() { + gfsnotify.RemoveCallback(c1.Id) + glog.Println("remove callback c1") + }) + // 10秒后移除c2的回调函数注册，所有的回调都移除，不再有任何打印信息输出 + gtime.SetTimeout(10*time.Second, func() { + gfsnotify.RemoveCallback(c2.Id) + glog.Println("remove callback c2") + }) + + select {} + +} \ No newline at end of file diff --git a/geg/os/gfsnotify/gfsnotify_callback_folder.go b/geg/os/gfsnotify/gfsnotify_callback_folder.go new file mode 100644 index 000000000..3d4148c04 --- /dev/null +++ b/geg/os/gfsnotify/gfsnotify_callback_folder.go @@ -0,0 +1,28 @@ +package main + +import ( + "gitee.com/johng/gf/g/os/gfsnotify" + "gitee.com/johng/gf/g/os/glog" + "gitee.com/johng/gf/g/os/gtime" + "time" +) + +func main() { + callback, err := gfsnotify.Add("/home/john/temp", func(event *gfsnotify.Event) { + glog.Println("callback") + }) + if err != nil { + panic(err) + } + + // 在此期间创建文件、目录、修改文件、删除文件 + + // 20秒后移除回调函数注册，所有的回调都移除，不再有任何打印信息输出 + gtime.SetTimeout(20*time.Second, func() { + gfsnotify.RemoveCallback(callback.Id) + glog.Println("remove callback") + }) + + select {} + +} \ No newline at end of file diff --git a/geg/other/test.go b/geg/other/test.go index aa0d5b0f5..0547cc693 100644 --- a/geg/other/test.go +++ b/geg/other/test.go @@ -2,12 +2,12 @@ package main import ( "fmt" - "gitee.com/johng/gf/g/os/gtime" - "strconv" + "strings" ) func main() { - fmt.Println(strconv.FormatInt(gtime.Nanosecond(), 32)) - fmt.Println(gtime.Second()) - fmt.Println(gtime.Nanosecond()) + fmt.Println(strings.Trim(` 1 `, "./- \n\r")) + //fmt.Println(math.MaxInt64) + //fmt.Println(gtime.Second()) + //fmt.Println(gtime.Nanosecond()) } \ No newline at end of file