tracker/udp/dispatcher.go

   1 package udp
   2
   3 import (
   4         "bytes"
   5         "fmt"
   6         "sync"
   7 )
   8
   9 // Maintains a mapping of transaction IDs to handlers.
  10 type Dispatcher struct {
  11         mu           sync.RWMutex
  12         transactions map[TransactionId]Transaction
  13 }
  14
  15 // The caller owns b.
  16 func (me *Dispatcher) Dispatch(b []byte) error {
  17         buf := bytes.NewBuffer(b)
  18         var rh ResponseHeader
  19         err := Read(buf, &rh)
  20         if err != nil {
  21                 return err
  22         }
  23         me.mu.RLock()
  24         defer me.mu.RUnlock()
  25         if t, ok := me.transactions[rh.TransactionId]; ok {
  26                 t.h(DispatchedResponse{
  27                         Header: rh,
  28                         Body:   append([]byte(nil), buf.Bytes()...),
  29                 })
  30                 return nil
  31         } else {
  32                 return fmt.Errorf("unknown transaction id %v", rh.TransactionId)
  33         }
  34 }
  35
  36 func (me *Dispatcher) forgetTransaction(id TransactionId) {
  37         me.mu.Lock()
  38         defer me.mu.Unlock()
  39         delete(me.transactions, id)
  40 }
  41
  42 func (me *Dispatcher) NewTransaction(h TransactionResponseHandler) Transaction {
  43         me.mu.Lock()
  44         defer me.mu.Unlock()
  45         for {
  46                 id := RandomTransactionId()
  47                 if _, ok := me.transactions[id]; ok {
  48                         continue
  49                 }
  50                 t := Transaction{
  51                         d:  me,
  52                         h:  h,
  53                         id: id,
  54                 }
  55                 if me.transactions == nil {
  56                         me.transactions = make(map[TransactionId]Transaction)
  57                 }
  58                 me.transactions[id] = t
  59                 return t
  60         }
  61 }
  62
  63 type DispatchedResponse struct {
  64         Header ResponseHeader
  65         Body   []byte
  66 }