- Timestamp:
- Nov 3, 2021, 2:17:16 PM (4 years ago)
- Location:
- trunk
- Files:
-
- 4 edited
-
downstream.go (modified) (7 diffs)
-
msgstore.go (modified) (2 diffs)
-
msgstore_fs.go (modified) (12 diffs)
-
msgstore_memory.go (modified) (1 diff)
Legend:
- Unmodified
- Added
- Removed
-
trunk/downstream.go
r664 r665 519 519 dc.SendMessage(msg) 520 520 521 if id == "" || !dc.messageSupports History(msg) {521 if id == "" || !dc.messageSupportsBacklog(msg) { 522 522 return 523 523 } … … 530 530 // isn't enabled. 531 531 func (dc *downstreamConn) advanceMessageWithID(msg *irc.Message, id string) { 532 if id == "" || !dc.messageSupports History(msg) {532 if id == "" || !dc.messageSupportsBacklog(msg) { 533 533 return 534 534 } … … 572 572 // marshalMessage re-formats a message coming from an upstream connection so 573 573 // that it's suitable for being sent on this downstream connection. Only 574 // messages that may appear in logs are supported, except MODE. 574 // messages that may appear in logs are supported, except MODE messages which 575 // may only appear in single-upstream mode. 575 576 func (dc *downstreamConn) marshalMessage(msg *irc.Message, net *network) *irc.Message { 577 if dc.network != nil { 578 return msg 579 } 580 576 581 msg = msg.Copy() 577 582 msg.Prefix = dc.marshalUserPrefix(net, msg.Prefix) … … 984 989 } 985 990 } 991 992 if dc.srv.LogPath != "" && dc.network != nil { 993 dc.setSupportedCap("draft/event-playback", "") 994 } else { 995 dc.unsetSupportedCap("draft/event-playback") 996 } 986 997 } 987 998 … … 1298 1309 } 1299 1310 1300 // messageSupports Historychecks whether the provided message can be sent as1311 // messageSupportsBacklog checks whether the provided message can be sent as 1301 1312 // part of an history batch. 1302 func (dc *downstreamConn) messageSupports History(msg *irc.Message) bool {1313 func (dc *downstreamConn) messageSupportsBacklog(msg *irc.Message) bool { 1303 1314 // Don't replay all messages, because that would mess up client 1304 1315 // state. For instance we just sent the list of users, sending 1305 1316 // PART messages for one of these users would be incorrect. 1306 // TODO: add support for draft/event-playback1307 1317 switch msg.Command { 1308 1318 case "PRIVMSG", "NOTICE": … … 1329 1339 dc.SendBatch("chathistory", []string{dc.marshalEntity(net, target)}, nil, func(batchRef irc.TagValue) { 1330 1340 for _, msg := range history { 1331 if !dc.messageSupportsHistory(msg) {1332 continue1333 }1334 1335 1341 if ch != nil && ch.Detached { 1336 1342 if net.detachedMessageNeedsRelay(ch, msg) { … … 2327 2333 } 2328 2334 2335 eventPlayback := dc.caps["draft/event-playback"] 2336 2329 2337 var history []*irc.Message 2330 2338 switch subcommand { 2331 2339 case "BEFORE": 2332 history, err = store.LoadBeforeTime(network, entity, bounds[0], time.Time{}, limit )2340 history, err = store.LoadBeforeTime(network, entity, bounds[0], time.Time{}, limit, eventPlayback) 2333 2341 case "AFTER": 2334 history, err = store.LoadAfterTime(network, entity, bounds[0], time.Now(), limit )2342 history, err = store.LoadAfterTime(network, entity, bounds[0], time.Now(), limit, eventPlayback) 2335 2343 case "BETWEEN": 2336 2344 if bounds[0].Before(bounds[1]) { 2337 history, err = store.LoadAfterTime(network, entity, bounds[0], bounds[1], limit )2345 history, err = store.LoadAfterTime(network, entity, bounds[0], bounds[1], limit, eventPlayback) 2338 2346 } else { 2339 history, err = store.LoadBeforeTime(network, entity, bounds[0], bounds[1], limit )2347 history, err = store.LoadBeforeTime(network, entity, bounds[0], bounds[1], limit, eventPlayback) 2340 2348 } 2341 2349 case "TARGETS": 2342 2350 // TODO: support TARGETS in multi-upstream mode 2343 targets, err := store.ListTargets(network, bounds[0], bounds[1], limit )2351 targets, err := store.ListTargets(network, bounds[0], bounds[1], limit, eventPlayback) 2344 2352 if err != nil { 2345 2353 dc.logger.Printf("failed fetching targets for chathistory: %v", err) -
trunk/msgstore.go
r549 r665 18 18 // used in history queries. 19 19 LastMsgID(network *network, entity string, t time.Time) (string, error) 20 // LoadLatestID queries the latest non-event messages for the given network, 21 // entity and date, up to a count of limit messages, sorted from oldest to newest. 20 22 LoadLatestID(network *network, entity, id string, limit int) ([]*irc.Message, error) 21 23 Append(network *network, entity string, msg *irc.Message) (id string, err error) … … 35 37 // It returns up to limit targets, starting from start and ending on end, 36 38 // both excluded. end may be before or after start. 37 ListTargets(network *network, start, end time.Time, limit int) ([]chatHistoryTarget, error) 39 // If events is false, only PRIVMSG/NOTICE messages are considered. 40 ListTargets(network *network, start, end time.Time, limit int, events bool) ([]chatHistoryTarget, error) 38 41 // LoadBeforeTime loads up to limit messages before start down to end. The 39 42 // returned messages must be between and excluding the provided bounds. 40 43 // end is before start. 41 LoadBeforeTime(network *network, entity string, start, end time.Time, limit int) ([]*irc.Message, error) 44 // If events is false, only PRIVMSG/NOTICE messages are considered. 45 LoadBeforeTime(network *network, entity string, start, end time.Time, limit int, events bool) ([]*irc.Message, error) 42 46 // LoadBeforeTime loads up to limit messages after start up to end. The 43 47 // returned messages must be between and excluding the provided bounds. 44 48 // end is after start. 45 LoadAfterTime(network *network, entity string, start, end time.Time, limit int) ([]*irc.Message, error) 49 // If events is false, only PRIVMSG/NOTICE messages are considered. 50 LoadAfterTime(network *network, entity string, start, end time.Time, limit int, events bool) ([]*irc.Message, error) 46 51 } 47 52 -
trunk/msgstore_fs.go
r644 r665 250 250 } 251 251 252 func parseMessage(line, entity string, ref time.Time ) (*irc.Message, time.Time, error) {252 func parseMessage(line, entity string, ref time.Time, events bool) (*irc.Message, time.Time, error) { 253 253 var hour, minute, second int 254 254 _, err := fmt.Sscanf(line, "[%02d:%02d:%02d] ", &hour, &minute, &second) … … 258 258 line = line[11:] 259 259 260 var cmd, sender, text string 261 if strings.HasPrefix(line, "<") { 262 cmd = "PRIVMSG" 263 parts := strings.SplitN(line[1:], "> ", 2) 260 var cmd string 261 var prefix *irc.Prefix 262 var params []string 263 if events && strings.HasPrefix(line, "*** ") { 264 parts := strings.SplitN(line[4:], " ", 2) 264 265 if len(parts) != 2 { 265 266 return nil, time.Time{}, nil 266 267 } 267 sender, text = parts[0], parts[1] 268 } else if strings.HasPrefix(line, "-") { 269 cmd = "NOTICE" 270 parts := strings.SplitN(line[1:], "- ", 2) 271 if len(parts) != 2 { 268 switch parts[0] { 269 case "Joins:", "Parts:", "Quits:": 270 args := strings.SplitN(parts[1], " ", 3) 271 if len(args) < 2 { 272 return nil, time.Time{}, nil 273 } 274 nick := args[0] 275 mask := strings.TrimSuffix(strings.TrimPrefix(args[1], "("), ")") 276 maskParts := strings.SplitN(mask, "@", 2) 277 if len(maskParts) != 2 { 278 return nil, time.Time{}, nil 279 } 280 prefix = &irc.Prefix{ 281 Name: nick, 282 User: maskParts[0], 283 Host: maskParts[1], 284 } 285 var reason string 286 if len(args) > 2 { 287 reason = strings.TrimSuffix(strings.TrimPrefix(args[2], "("), ")") 288 } 289 switch parts[0] { 290 case "Joins:": 291 cmd = "JOIN" 292 params = []string{entity} 293 case "Parts:": 294 cmd = "PART" 295 if reason != "" { 296 params = []string{entity, reason} 297 } else { 298 params = []string{entity} 299 } 300 case "Quits:": 301 cmd = "QUIT" 302 if reason != "" { 303 params = []string{reason} 304 } 305 } 306 default: 307 nick := parts[0] 308 rem := parts[1] 309 if r := strings.TrimPrefix(rem, "is now known as "); r != rem { 310 cmd = "NICK" 311 prefix = &irc.Prefix{ 312 Name: nick, 313 } 314 params = []string{r} 315 } else if r := strings.TrimPrefix(rem, "was kicked by "); r != rem { 316 args := strings.SplitN(r, " ", 2) 317 if len(args) != 2 { 318 return nil, time.Time{}, nil 319 } 320 cmd = "KICK" 321 prefix = &irc.Prefix{ 322 Name: args[0], 323 } 324 reason := strings.TrimSuffix(strings.TrimPrefix(args[1], "("), ")") 325 params = []string{entity, nick} 326 if reason != "" { 327 params = append(params, reason) 328 } 329 } else if r := strings.TrimPrefix(rem, "changes topic to "); r != rem { 330 cmd = "TOPIC" 331 prefix = &irc.Prefix{ 332 Name: nick, 333 } 334 topic := strings.TrimSuffix(strings.TrimPrefix(r, "'"), "'") 335 params = []string{entity, topic} 336 } else if r := strings.TrimPrefix(rem, "sets mode: "); r != rem { 337 cmd = "MODE" 338 prefix = &irc.Prefix{ 339 Name: nick, 340 } 341 params = append([]string{entity}, strings.Split(r, " ")...) 342 } else { 343 return nil, time.Time{}, nil 344 } 345 } 346 } else { 347 var sender, text string 348 if strings.HasPrefix(line, "<") { 349 cmd = "PRIVMSG" 350 parts := strings.SplitN(line[1:], "> ", 2) 351 if len(parts) != 2 { 352 return nil, time.Time{}, nil 353 } 354 sender, text = parts[0], parts[1] 355 } else if strings.HasPrefix(line, "-") { 356 cmd = "NOTICE" 357 parts := strings.SplitN(line[1:], "- ", 2) 358 if len(parts) != 2 { 359 return nil, time.Time{}, nil 360 } 361 sender, text = parts[0], parts[1] 362 } else if strings.HasPrefix(line, "* ") { 363 cmd = "PRIVMSG" 364 parts := strings.SplitN(line[2:], " ", 2) 365 if len(parts) != 2 { 366 return nil, time.Time{}, nil 367 } 368 sender, text = parts[0], "\x01ACTION "+parts[1]+"\x01" 369 } else { 272 370 return nil, time.Time{}, nil 273 371 } 274 sender, text = parts[0], parts[1] 275 } else if strings.HasPrefix(line, "* ") { 276 cmd = "PRIVMSG" 277 parts := strings.SplitN(line[2:], " ", 2) 278 if len(parts) != 2 { 279 return nil, time.Time{}, nil 280 } 281 sender, text = parts[0], "\x01ACTION "+parts[1]+"\x01" 282 } else { 283 return nil, time.Time{}, nil 372 373 prefix = &irc.Prefix{Name: sender} 374 params = []string{entity, text} 284 375 } 285 376 … … 291 382 "time": irc.TagValue(t.UTC().Format(serverTimeLayout)), 292 383 }, 293 Prefix: &irc.Prefix{Name: sender},384 Prefix: prefix, 294 385 Command: cmd, 295 Params: []string{entity, text},386 Params: params, 296 387 } 297 388 return msg, t, nil 298 389 } 299 390 300 func (ms *fsMessageStore) parseMessagesBefore(network *network, entity string, ref time.Time, end time.Time, limit int, afterOffset int64) ([]*irc.Message, error) {391 func (ms *fsMessageStore) parseMessagesBefore(network *network, entity string, ref time.Time, end time.Time, events bool, limit int, afterOffset int64) ([]*irc.Message, error) { 301 392 path := ms.logPath(network, entity, ref) 302 393 f, err := os.Open(path) … … 322 413 323 414 for sc.Scan() { 324 msg, t, err := parseMessage(sc.Text(), entity, ref )415 msg, t, err := parseMessage(sc.Text(), entity, ref, events) 325 416 if err != nil { 326 417 return nil, err … … 354 445 } 355 446 356 func (ms *fsMessageStore) parseMessagesAfter(network *network, entity string, ref time.Time, end time.Time, limit int) ([]*irc.Message, error) {447 func (ms *fsMessageStore) parseMessagesAfter(network *network, entity string, ref time.Time, end time.Time, events bool, limit int) ([]*irc.Message, error) { 357 448 path := ms.logPath(network, entity, ref) 358 449 f, err := os.Open(path) … … 368 459 sc := bufio.NewScanner(f) 369 460 for sc.Scan() && len(history) < limit { 370 msg, t, err := parseMessage(sc.Text(), entity, ref )461 msg, t, err := parseMessage(sc.Text(), entity, ref, events) 371 462 if err != nil { 372 463 return nil, err … … 386 477 } 387 478 388 func (ms *fsMessageStore) LoadBeforeTime(network *network, entity string, start time.Time, end time.Time, limit int ) ([]*irc.Message, error) {479 func (ms *fsMessageStore) LoadBeforeTime(network *network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) { 389 480 start = start.In(time.Local) 390 481 end = end.In(time.Local) … … 393 484 tries := 0 394 485 for remaining > 0 && tries < fsMessageStoreMaxTries && end.Before(start) { 395 buf, err := ms.parseMessagesBefore(network, entity, start, end, remaining, -1)486 buf, err := ms.parseMessagesBefore(network, entity, start, end, events, remaining, -1) 396 487 if err != nil { 397 488 return nil, err … … 411 502 } 412 503 413 func (ms *fsMessageStore) LoadAfterTime(network *network, entity string, start time.Time, end time.Time, limit int ) ([]*irc.Message, error) {504 func (ms *fsMessageStore) LoadAfterTime(network *network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) { 414 505 start = start.In(time.Local) 415 506 end = end.In(time.Local) … … 418 509 tries := 0 419 510 for remaining > 0 && tries < fsMessageStoreMaxTries && start.Before(end) { 420 buf, err := ms.parseMessagesAfter(network, entity, start, end, remaining)511 buf, err := ms.parseMessagesAfter(network, entity, start, end, events, remaining) 421 512 if err != nil { 422 513 return nil, err … … 461 552 } 462 553 463 buf, err := ms.parseMessagesBefore(network, entity, t, time.Time{}, remaining, offset)554 buf, err := ms.parseMessagesBefore(network, entity, t, time.Time{}, false, remaining, offset) 464 555 if err != nil { 465 556 return nil, err … … 479 570 } 480 571 481 func (ms *fsMessageStore) ListTargets(network *network, start, end time.Time, limit int ) ([]chatHistoryTarget, error) {572 func (ms *fsMessageStore) ListTargets(network *network, start, end time.Time, limit int, events bool) ([]chatHistoryTarget, error) { 482 573 start = start.In(time.Local) 483 574 end = end.In(time.Local) -
trunk/msgstore_memory.go
r517 r665 75 75 76 76 func (ms *memoryMessageStore) Append(network *network, entity string, msg *irc.Message) (string, error) { 77 switch msg.Command { 78 case "PRIVMSG", "NOTICE": 79 default: 80 return "", nil 81 } 82 77 83 k := ringBufferKey{networkID: network.ID, entity: entity} 78 84 rb, ok := ms.buffers[k]
Note:
See TracChangeset
for help on using the changeset viewer.
