Class: StarlingServer::Handler

Inherits:
EventMachine::Connection
  • Object
show all
Defined in:
lib/starling/handler.rb

Overview

This is an internal class that’s used by Starling::Server to handle the MemCache protocol and act as an interface between the Server and the QueueCollection.

Constant Summary collapse

DATA_PACK_FMT = 
"Ia*".freeze
ERR_UNKNOWN_COMMAND =

ERROR responses

"CLIENT_ERROR bad command line format\r\n".freeze
CMD_RE = 
/\A(g)et\s+([!-~]{1,250})\s{0,}\r\n\Z|\A(s)et\s+([!-~]{1,250})\s+(\d+)\s+(\d+)\s+(\d+)\s{0,}(noreply)?\s{0,}\r\n\Z|\A(d)elete\s+([!-~]{1,250})\s+(\d+)\s{0,}\r\n\Z|\As(t)ats\s{0,}\r\n\Z|\As(h)utdown\s{0,}\r\n\Z|\A(q)uit\s{0,}\r\n\Z/.freeze
GET_RESPONSE =

GET Responses

"VALUE %s %s %s\r\n%s\r\nEND\r\n".freeze
GET_RESPONSE_EMPTY = 
"END\r\n".freeze
SET_RESPONSE_SUCCESS =

SET Responses

"STORED\r\n".freeze
SET_RESPONSE_FAILURE = 
"NOT STORED\r\n".freeze
SET_CLIENT_DATA_ERROR = 
"CLIENT_ERROR bad data chunk\r\nERROR\r\n".freeze
DELETE_RESPONSE =

DELETE Responses

"END\r\n".freeze
STATS_RESPONSE =

STAT Response

"STAT pid %d\r
STAT uptime %d\r
STAT time %d\r
STAT version %s\r
STAT rusage_user %0.6f\r
STAT rusage_system %0.6f\r
STAT curr_items %d\r
STAT total_items %d\r
STAT bytes %d\r
STAT curr_connections %d\r
STAT total_connections %d\r
STAT cmd_get %d\r
STAT cmd_set %d\r
STAT get_hits %d\r
STAT get_misses %d\r
STAT bytes_read %d\r
STAT bytes_written %d\r
STAT limit_maxbytes %d\r
%sEND\r\n".freeze
QUEUE_STATS_RESPONSE = 
"STAT queue_%s_items %d\r
STAT queue_%s_total_items %d\r
STAT queue_%s_logsize %d\r
STAT queue_%s_expired_items %d\r
STAT queue_%s_age %d\r\n".freeze
@@next_session_id = 
1

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Handler

Creates a new handler for the MemCache protocol that communicates with a given client.



69
70
71
 
# File 'lib/starling/handler.rb', line 69

def initialize(options = {})
  @opts = options
end

Instance Method Details

#post_init ⇒ Object

Process incoming commands from the attached client.



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/starling/handler.rb', line 76

def post_init
  @stash = []
  @data = ""
  @data_buf = ""
  @server = @opts[:server]
  @logger = StarlingServer::Base.logger
  @expiry_stats = Hash.new(0)
  @expected_length = nil
  @server.stats[:total_connections] += 1
  set_comm_inactivity_timeout @opts[:timeout]
  @queue_collection = @opts[:queue]

  @session_id = @@next_session_id
  @@next_session_id += 1

  peer = Socket.unpack_sockaddr_in(get_peername)
  #@logger.debug "(#{@session_id}) New session from #{peer[1]}:#{peer[0]}"
end

#process(data) ⇒ Object 



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/starling/handler.rb', line 107

def process(data)
  data = @data_buf + data unless @data_buf.size == 0
  if @expected_length && data.size == @expected_length
    @data_buf = ""
    return  set_data(data)
  elsif @expected_length
    @data_buf = data
    return
  end
  m = CMD_RE.match(data)
  if ! m.nil?
    args = m.captures.compact
    case args[0][0]
    when ?s #set command
      @server.stats[:set_requests] += 1
      set(args[1], args[2], args[3], args[4].to_i)
    when ?g #get command
      @server.stats[:get_requests] += 1
      get(args[1])
    when ?t # stats command
      stats
    when ?h # shutdown command
      Runner::shutdown
    when ?d # delete command
      delete args[1]
    when ?q # quit command
      # ignore the command, client is closing connection.
      return nil
    else
      logger.warn "Unknown command (in case): #{data}."
      respond ERR_UNKNOWN_COMMAND
    end
  else
    logger.warn "Unknown command: #{data}."
    respond ERR_UNKNOWN_COMMAND
  end
end

#receive_data(incoming) ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/starling/handler.rb', line 95

def receive_data(incoming)
  @server.stats[:bytes_read] += incoming.size
  @data << incoming

  while ind = @data.index("\r\n")
    to_proc = @data[0,ind+2]
    @data = @data[ind+2,@data.size]
    response = process(to_proc)
    send_data response if response
  end
end

#unbind ⇒ Object 



145
146
147
 
# File 'lib/starling/handler.rb', line 145

def unbind
  #@logger.debug "(#{@session_id}) connection ends"
end