Class: AvroTurf::ConfluentSchemaRegistry

Inherits:
Object
  • Object
show all
Defined in:
lib/avro_turf/confluent_schema_registry.rb

Constant Summary collapse

CONTENT_TYPE =
"application/vnd.schemaregistry.v1+json".freeze

Instance Method Summary collapse

Constructor Details

#initialize(url, logger: Logger.new($stdout), proxy: nil, user: nil, password: nil, client_cert: nil, client_key: nil, client_key_pass: nil, client_cert_data: nil, client_key_data: nil) ⇒ ConfluentSchemaRegistry

Returns a new instance of ConfluentSchemaRegistry.


6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# File 'lib/avro_turf/confluent_schema_registry.rb', line 6

def initialize(
  url,
  logger: Logger.new($stdout),
  proxy: nil,
  user: nil,
  password: nil,
  client_cert: nil,
  client_key: nil,
  client_key_pass: nil,
  client_cert_data: nil,
  client_key_data: nil
)
  @logger = logger
  headers = {
    "Content-Type" => CONTENT_TYPE
  }
  headers[:proxy] = proxy unless proxy.nil?
  @connection = Excon.new(
    url,
    headers: headers,
    user: user,
    password: password,
    client_cert: client_cert,
    client_key: client_key,
    client_key_pass: client_key_pass,
    client_cert_data: client_cert_data,
    client_key_data: client_key_data
  )
end

Instance Method Details

#check(subject, schema) ⇒ Object

Check if a schema exists. Returns nil if not found.


68
69
70
71
72
73
# File 'lib/avro_turf/confluent_schema_registry.rb', line 68

def check(subject, schema)
  data = post("/subjects/#{subject}",
              expects: [200, 404],
              body: { schema: schema.to_s }.to_json)
  data unless data.has_key?("error_code")
end

#compatible?(subject, schema, version = 'latest') ⇒ Boolean

Check if a schema is compatible with the stored version. Returns:

  • true if compatible

  • nil if the subject or version does not exist

  • false if incompatible

docs.confluent.io/3.1.2/schema-registry/docs/api.html#compatibility

Returns:

  • (Boolean)

81
82
83
84
85
# File 'lib/avro_turf/confluent_schema_registry.rb', line 81

def compatible?(subject, schema, version = 'latest')
  data = post("/compatibility/subjects/#{subject}/versions/#{version}",
              expects: [200, 404], body: { schema: schema.to_s }.to_json)
  data.fetch('is_compatible', false) unless data.has_key?('error_code')
end

#fetch(id) ⇒ Object


36
37
38
39
40
# File 'lib/avro_turf/confluent_schema_registry.rb', line 36

def fetch(id)
  @logger.info "Fetching schema with id #{id}"
  data = get("/schemas/ids/#{id}")
  data.fetch("schema")
end

#global_configObject

Get global config


88
89
90
# File 'lib/avro_turf/confluent_schema_registry.rb', line 88

def global_config
  get("/config")
end

#register(subject, schema) ⇒ Object


42
43
44
45
46
47
48
49
50
# File 'lib/avro_turf/confluent_schema_registry.rb', line 42

def register(subject, schema)
  data = post("/subjects/#{subject}/versions", body: { schema: schema.to_s }.to_json)

  id = data.fetch("id")

  @logger.info "Registered schema for subject `#{subject}`; id = #{id}"

  id
end

#subject_config(subject) ⇒ Object

Get config for subject


98
99
100
# File 'lib/avro_turf/confluent_schema_registry.rb', line 98

def subject_config(subject)
  get("/config/#{subject}")
end

#subject_version(subject, version = 'latest') ⇒ Object

Get a specific version for a subject


63
64
65
# File 'lib/avro_turf/confluent_schema_registry.rb', line 63

def subject_version(subject, version = 'latest')
  get("/subjects/#{subject}/versions/#{version}")
end

#subject_versions(subject) ⇒ Object

List all versions for a subject


58
59
60
# File 'lib/avro_turf/confluent_schema_registry.rb', line 58

def subject_versions(subject)
  get("/subjects/#{subject}/versions")
end

#subjectsObject

List all subjects


53
54
55
# File 'lib/avro_turf/confluent_schema_registry.rb', line 53

def subjects
  get('/subjects')
end

#update_global_config(config) ⇒ Object

Update global config


93
94
95
# File 'lib/avro_turf/confluent_schema_registry.rb', line 93

def update_global_config(config)
  put("/config", body: config.to_json)
end

#update_subject_config(subject, config) ⇒ Object

Update config for subject


103
104
105
# File 'lib/avro_turf/confluent_schema_registry.rb', line 103

def update_subject_config(subject, config)
  put("/config/#{subject}", body: config.to_json)
end