Class: Fluent::BigQueryPlugin::LoadRequestBodyWrapper

Inherits:
Object
  • Object
show all
Defined in:
lib/fluent/plugin/bigquery/load_request_body_wrapper.rb

Constant Summary collapse

JSON_PRETTY_DUMP =

https://developers.google.com/bigquery/loading-data-into-bigquery#loaddatapostrequest

JSON::State.new(space: " ", indent:"  ", object_nl:"\n", array_nl:"\n")
CONTENT_TYPE_FIRST = 
"Content-Type: application/json; charset=UTF-8\n\n"
CONTENT_TYPE_SECOND = 
"Content-Type: application/octet-stream\n\n"
MULTIPART_BOUNDARY = 
"--xxx\n"
MULTIPART_BOUNDARY_END = 
"--xxx--\n"

Instance Method Summary collapse

Constructor Details

#initialize(project_id, dataset_id, table_id, field_defs, buffer) ⇒ LoadRequestBodyWrapper

Returns a new instance of LoadRequestBodyWrapper.



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/fluent/plugin/bigquery/load_request_body_wrapper.rb', line 29

def initialize(project_id, dataset_id, table_id, field_defs, buffer)
  @metadata = {
    configuration: {
      load: {
        sourceFormat: "<required for JSON files>",
        schema: {
          fields: field_defs
        },
        destinationTable: {
          projectId: project_id,
          datasetId: dataset_id,
          tableId: table_id
        }
      }
    }
  }

  @non_buffer = MULTIPART_BOUNDARY + CONTENT_TYPE_FIRST + @metadata.to_json(JSON_PRETTY_DUMP) + "\n" +
    MULTIPART_BOUNDARY + CONTENT_TYPE_SECOND
  @non_buffer.force_encoding("ASCII-8BIT")
  @non_buffer_bytesize = @non_buffer.bytesize

  @buffer = buffer # read
  @buffer_bytesize = @buffer.size # Fluentd Buffer Chunk #size -> bytesize

  @footer = MULTIPART_BOUNDARY_END.force_encoding("ASCII-8BIT")

  @contents_bytesize = @non_buffer_bytesize + @buffer_bytesize
  @total_bytesize = @contents_bytesize + MULTIPART_BOUNDARY_END.bytesize

  @whole_data = nil

  @counter = 0
  @eof = false
end

Instance Method Details

#eof? ⇒ Boolean

Returns:

  • (Boolean) 


98
99
100
 
# File 'lib/fluent/plugin/bigquery/load_request_body_wrapper.rb', line 98

def eof?
  @eof
end

#read(length = nil, outbuf = "") ⇒ Object

Raises:

  • (ArgumentError) 


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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/fluent/plugin/bigquery/load_request_body_wrapper.rb', line 107

def read(length=nil, outbuf="")
  raise ArgumentError, "negative read length" if length && length < 0
  return (length.nil? || length == 0) ? "" : nil if @eof
  return outbuf if length == 0

  # read all data
  if length.nil? || length >= @total_bytesize
    @whole_data ||= @buffer.read.force_encoding("ASCII-8BIT")

    if @counter.zero?
      outbuf.replace(@non_buffer)
      outbuf << @whole_data
      outbuf << @footer
    elsif @counter < @non_buffer_bytesize
      outbuf.replace(@non_buffer[ @counter .. -1 ])
      outbuf << @whole_data
      outbuf << @footer
    elsif @counter < @contents_bytesize
      outbuf.replace(@whole_data[ (@counter - @non_buffer_bytesize) .. -1 ])
      outbuf << @footer
    else
      outbuf.replace(@footer[ (@counter - @contents_bytesize) .. -1 ])
    end
    @counter = @total_bytesize
    @eof = true
    return outbuf
  end

  # In ruby script level (non-ext module), we cannot prevent to change outbuf length or object re-assignment
  outbuf.replace("")

  # return first part (metadata)
  if @counter < @non_buffer_bytesize
    non_buffer_part = @non_buffer[@counter, length]
    if non_buffer_part
      outbuf << non_buffer_part
      length -= non_buffer_part.bytesize
      @counter += non_buffer_part.bytesize
    end
  end
  return outbuf if length < 1

  # return second part (buffer content)
  if @counter < @contents_bytesize
    @whole_data ||= @buffer.read.force_encoding("ASCII-8BIT")
    buffer_part = @whole_data[@counter - @non_buffer_bytesize, length]
    if buffer_part
      outbuf << buffer_part
      length -= buffer_part.bytesize
      @counter += buffer_part.bytesize
    end
  end
  return outbuf if length < 1

  # return footer
  footer_part = @footer[@counter - @contents_bytesize, length]
  if footer_part
    outbuf << footer_part
    @counter += footer_part.bytesize
    @eof = true if @counter >= @total_bytesize
  end

  outbuf
end

#rewind ⇒ Object

sample_body = <<EOF --xxx Content-Type: application/json; charset=UTF-8

{ "configuration": { "load": { "sourceFormat": "", "schema": { "fields": [ "type":"STRING", "type":"INTEGER" ] }, "destinationTable": { "projectId": "projectId", "datasetId": "datasetId", "tableId": "tableId" } } } } --xxx Content-Type: application/octet-stream

--xxx-- EOF 


93
94
95
96
 
# File 'lib/fluent/plugin/bigquery/load_request_body_wrapper.rb', line 93

def rewind
  @counter = 0
  @eof = false
end

#to_str ⇒ Object 



102
103
104
105
 
# File 'lib/fluent/plugin/bigquery/load_request_body_wrapper.rb', line 102

def to_str
  rewind
  self.read # all data
end