udp.go

   1 package anidb
   2
   3 import (
   4         "encoding/gob"
   5         "github.com/Kovensky/go-anidb/udp"
   6         "log"
   7         "sync"
   8         "time"
   9 )
  10
  11 func init() {
  12         gob.RegisterName("*github.com/Kovensky/go-anidb.banCache", &banCache{})
  13 }
  14
  15 const banDuration = 30*time.Minute + 1*time.Second
  16
  17 type banCache struct{ time.Time }
  18
  19 func (c *banCache) Touch() {
  20         c.Time = time.Now()
  21 }
  22 func (c *banCache) IsStale() bool {
  23         return time.Now().Sub(c.Time) > banDuration
  24 }
  25
  26 // Returns whether the last UDP API access returned a 555 BANNED message.
  27 func Banned() bool {
  28         var banTime banCache
  29         cache.Get(&banTime, "banned")
  30
  31         stale := banTime.IsStale()
  32         if stale {
  33                 cache.Delete("banned")
  34         }
  35         return !stale
  36 }
  37
  38 func setBanned() {
  39         cache.Set(&banCache{}, "banned")
  40 }
  41
  42 type paramSet struct {
  43         cmd    string
  44         params paramMap
  45         ch     chan udpapi.APIReply
  46 }
  47
  48 type udpWrap struct {
  49         *udpapi.AniDBUDP
  50
  51         sendLock    sync.Mutex
  52         sendQueueCh chan paramSet
  53
  54         credLock    sync.Mutex
  55         credentials *credentials
  56         connected   bool
  57 }
  58
  59 func newUDPWrap() *udpWrap {
  60         u := &udpWrap{
  61                 AniDBUDP:    udpapi.NewAniDBUDP(),
  62                 sendQueueCh: make(chan paramSet, 10),
  63         }
  64         go u.sendQueue()
  65         return u
  66 }
  67
  68 type paramMap udpapi.ParamMap // shortcut
  69
  70 type noauthAPIReply struct {
  71         udpapi.APIReply
  72 }
  73
  74 func (r *noauthAPIReply) Code() int {
  75         return 501
  76 }
  77
  78 func (r *noauthAPIReply) Text() string {
  79         return "LOGIN FIRST"
  80 }
  81
  82 func (r *noauthAPIReply) Error() error {
  83         return &udpapi.APIError{Code: r.Code(), Desc: r.Text()}
  84 }
  85
  86 type bannedAPIReply struct {
  87         udpapi.APIReply
  88 }
  89
  90 func (r *bannedAPIReply) Code() int {
  91         return 555
  92 }
  93 func (r *bannedAPIReply) Text() string {
  94         return "BANNED"
  95 }
  96 func (r *bannedAPIReply) Error() error {
  97         return &udpapi.APIError{Code: r.Code(), Desc: r.Text()}
  98 }
  99
 100 var bannedReply udpapi.APIReply = &bannedAPIReply{}
 101
 102 func logRequest(set paramSet) {
 103         switch set.cmd {
 104         case "AUTH":
 105                 log.Printf("UDP>>> AUTH user=%s\n", set.params["user"])
 106         default:
 107                 log.Printf("UDP>>> %s %s\n", set.cmd, udpapi.ParamMap(set.params).String())
 108         }
 109 }
 110
 111 func logReply(reply udpapi.APIReply) {
 112         log.Printf("UDP<<< %d %s\n", reply.Code(), reply.Text())
 113 }
 114
 115 func (udp *udpWrap) sendQueue() {
 116         initialWait := 5 * time.Second
 117         wait := initialWait
 118         for set := range udp.sendQueueCh {
 119         Retry:
 120                 if Banned() {
 121                         set.ch <- bannedReply
 122                         close(set.ch)
 123                         continue
 124                 }
 125
 126                 logRequest(set)
 127                 reply := <-udp.AniDBUDP.SendRecv(set.cmd, udpapi.ParamMap(set.params))
 128
 129                 if reply.Error() == udpapi.TimeoutError {
 130                         // retry
 131                         wait = wait * 2
 132                         if wait > time.Minute {
 133                                 wait = time.Minute
 134                         }
 135                         log.Printf("UDP--- Timeout; waiting %s before retry", wait)
 136
 137                         delete(set.params, "s")
 138                         delete(set.params, "tag")
 139
 140                         time.Sleep(wait)
 141                         goto Retry
 142                 }
 143                 logReply(reply)
 144
 145                 wait = initialWait
 146
 147                 switch reply.Code() {
 148                 case 403, 501, 506: // not logged in, or session expired
 149                         if r := udp.ReAuth(); r.Error() == nil {
 150                                 // retry
 151
 152                                 delete(set.params, "s")
 153                                 delete(set.params, "tag")
 154
 155                                 goto Retry
 156                         }
 157                 case 503, 504: // client library rejected
 158                         panic(reply.Error())
 159                 // 555: IP (and user, possibly client) temporarily banned
 160                 // 601: Server down (treat the same as a ban)
 161                 case 555, 601:
 162                         setBanned()
 163                 }
 164                 set.ch <- reply
 165                 close(set.ch)
 166         }
 167 }
 168
 169 func (udp *udpWrap) SendRecv(cmd string, params paramMap) <-chan udpapi.APIReply {
 170         ch := make(chan udpapi.APIReply, 1)
 171
 172         udp.sendLock.Lock()
 173         defer udp.sendLock.Unlock()
 174
 175         if Banned() {
 176                 ch <- bannedReply
 177                 close(ch)
 178                 return ch
 179         }
 180
 181         if !udp.connected {
 182                 if r := udp.ReAuth(); r.Error() != nil {
 183                         ch <- r
 184                         close(ch)
 185                         return ch
 186                 }
 187         }
 188
 189         udp.sendQueueCh <- paramSet{
 190                 cmd:    cmd,
 191                 params: params,
 192                 ch:     ch,
 193         }
 194
 195         return ch
 196 }