Class: NusaDB::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/nusadb/connection.rb

Overview

One authenticated connection to a NusaDB server.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(host: '127.0.0.1', port: 5678, user: 'nusa-root', database: 'nusadb', password: 'nusa-root') ⇒ Connection

Returns a new instance of Connection.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/nusadb/connection.rb', line 65

def initialize(host: '127.0.0.1', port: 5678, user: 'nusa-root', database: 'nusadb', password: 'nusa-root')
  @user = user
  @counter = 0
  @backend_key = nil
  # Async LISTEN/NOTIFY messages received while reading other responses; drained by #notifications.
  @notifications = []
  begin
    @socket = TCPSocket.new(host, port)
    # Disable Nagle's algorithm: this is a request/response protocol, so
    # coalescing small frames only adds delayed-ACK latency per round-trip.
    @socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
  rescue SystemCallError => e
    raise NusaError.new("nusadb: connect failed: #{e.message}", '08001')
  end
  handshake(user, database, password)
end

Instance Attribute Details

#backend_key ⇒ Object (readonly)

Returns the value of attribute backend_key.



63
64
65
 
# File 'lib/nusadb/connection.rb', line 63

def backend_key
  @backend_key
end

Instance Method Details

#close ⇒ Object 



163
164
165
166
167
168
169
 
# File 'lib/nusadb/connection.rb', line 163

def close
  send_frame(Protocol.terminate)
rescue IOError, SystemCallError
  # best effort
ensure
  @socket.close unless @socket.closed?
end

#copy_in(sql, source) ⇒ Object

Bulk-load via COPY ... FROM STDIN (§12.1). source is an IO (responds to #read) or a String already in the server's text format (tab-delimited, \N for NULL). Returns the loaded row count. A refused COPY (bad SQL, an RLS-protected table) raises; the connection stays usable.



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
 
# File 'lib/nusadb/connection.rb', line 118

def copy_in(sql, source)
  send_frame(Protocol.query(sql))
  await_copy_start(Protocol::B_COPY_IN)
  if source.respond_to?(:read)
    loop do
      # A read error on the source aborts the load with CopyFail (mirrors the other drivers),
      # then drains to ReadyForQuery so the connection stays usable.
      begin
        chunk = source.read(65_536)
      rescue StandardError => e
        send_frame(Protocol.copy_fail('nusadb: client read error during COPY'))
        drain_to_ready
        raise NusaError.new("nusadb: COPY source read failed: #{e.message}", '57014')
      end
      break if chunk.nil?

      send_frame(Protocol.copy_data(chunk.b)) unless chunk.empty?
    end
  else
    bytes = source.to_s.dup.force_encoding(Encoding::BINARY)
    (0...bytes.bytesize).step(65_536) { |off| send_frame(Protocol.copy_data(bytes.byteslice(off, 65_536))) }
  end
  send_frame(Protocol.copy_done)
  finish_copy
end

#copy_out(sql, sink) ⇒ Object

Bulk-export via COPY ... TO STDOUT (§12.2). Writes each data chunk to sink (responds to #write) in the server's text format; returns the exported row count.



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/nusadb/connection.rb', line 146

def copy_out(sql, sink)
  send_frame(Protocol.query(sql))
  await_copy_start(Protocol::B_COPY_OUT)
  loop do
    type, reader = read_message
    case type
    when Protocol::B_COPY_DATA then sink.write(reader.rest)
    when Protocol::B_COPY_DONE then break
    when Protocol::B_ERROR
      err = server_error(reader)
      drain_to_ready
      raise err
    end
  end
  finish_copy
end

#execute(sql, params = []) ⇒ Object

Run a statement returning the affected-row count.



102
103
104
 
# File 'lib/nusadb/connection.rb', line 102

def execute(sql, params = [])
  query(sql, params).affected
end

#execute_many(sql, param_sets) ⇒ Object

Run one statement once per parameter set, reusing a single prepared statement — the bulk insert/update path. Returns an Array of per-set affected-row counts. One parse, then one execute per set (the wire protocol has no batch pipeline, so this is N round-trips, not one); the first failing set raises.



110
111
112
113
 
# File 'lib/nusadb/connection.rb', line 110

def execute_many(sql, param_sets)
  stmt = prepare(sql)
  (param_sets || []).map { |params| stmt.execute(params).affected }
end

#listen(channel) ⇒ Object

Subscribe to asynchronous notifications on channel (+LISTEN channel+). Collect them with #notifications or #poll.



173
174
175
176
 
# File 'lib/nusadb/connection.rb', line 173

def listen(channel)
  run_simple("LISTEN #{quote_identifier(channel)}")
  nil
end

#notifications ⇒ Object

Return and clear the notifications already received (buffered while reading other responses). Does not touch the socket -- use #poll to wait for new ones.



195
196
197
198
199
 
# File 'lib/nusadb/connection.rb', line 195

def notifications
  pending = @notifications
  @notifications = []
  pending
end

#notify(channel, payload = nil) ⇒ Object

Send a notification on channel with an optional payload (+NOTIFY channel[, 'payload']+).



186
187
188
189
190
191
 
# File 'lib/nusadb/connection.rb', line 186

def notify(channel, payload = nil)
  sql = "NOTIFY #{quote_identifier(channel)}"
  sql += ", #{quote_literal(payload)}" unless payload.nil?
  run_simple(sql)
  nil
end

#poll(timeout = 0) ⇒ Object

Wait up to timeout seconds for the next notification (a buffered one is returned immediately; nil on timeout). timeout of nil blocks until one arrives. Only meaningful after #listen.



203
204
205
206
207
208
209
210
211
212
213
214
215
 
# File 'lib/nusadb/connection.rb', line 203

def poll(timeout = 0)
  return @notifications.shift unless @notifications.empty?

  return nil if IO.select([@socket], nil, nil, timeout).nil?

  type, reader = read_message
  case type
  when Protocol::B_NOTIFICATION then decode_notification(reader)
  when Protocol::B_ERROR then raise server_error(reader)
  else
    raise NusaError.new('nusadb: unexpected message while polling for notifications', '08006')
  end
end

#prepare(sql) ⇒ Object

Prepare a statement for repeated execution; returns an object with #execute(params).



93
94
95
96
97
98
99
 
# File 'lib/nusadb/connection.rb', line 93

def prepare(sql)
  name = fresh_name('stmt')
  send_frame(Protocol.parse(name, sql))
  send_frame(Protocol.sync)
  collect # surfaces a parse error
  Statement.new(self, name)
end

#query(sql, params = []) ⇒ Object

Run one SQL string, optionally with positional parameters ($1, $2, …).



83
84
85
86
87
88
89
90
 
# File 'lib/nusadb/connection.rb', line 83

def query(sql, params = [])
  raw = if params && !params.empty?
          run_extended(sql, params)
        else
          run_simple(sql)
        end
  shape(raw)
end

#run_prepared(name, params) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



218
219
220
221
222
223
224
225
 
# File 'lib/nusadb/connection.rb', line 218

def run_prepared(name, params)
  portal = fresh_name('portal')
  send_frame(Protocol.bind(portal, name, encode_params(params)))
  send_frame(Protocol.describe_portal(portal))
  send_frame(Protocol.execute(portal, 0))
  send_frame(Protocol.sync)
  shape(collect)
end

#unlisten(channel = nil) ⇒ Object

Stop listening on channel (+UNLISTEN channel+); nil unlistens all (+UNLISTEN *+).



179
180
181
182
183
 
# File 'lib/nusadb/connection.rb', line 179

def unlisten(channel = nil)
  target = channel.nil? ? '*' : quote_identifier(channel)
  run_simple("UNLISTEN #{target}")
  nil
end