{"action":"created","issue":{"url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/568","repository_url":"https://api.github.com/repos/zendesk/ruby-kafka","labels_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/568/labels{/name}","comments_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/568/comments","events_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/568/events","html_url":"https://github.com/zendesk/ruby-kafka/issues/568","id":315594066,"number":568,"title":"Persistent offline producer cache?","user":{"login":"rgaufman","id":33842342,"avatar_url":"https://avatars3.githubusercontent.com/u/33842342?v=4","gravatar_id":"","url":"https://api.github.com/users/rgaufman","html_url":"https://github.com/rgaufman","followers_url":"https://api.github.com/users/rgaufman/followers","following_url":"https://api.github.com/users/rgaufman/following{/other_user}","gists_url":"https://api.github.com/users/rgaufman/gists{/gist_id}","starred_url":"https://api.github.com/users/rgaufman/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/rgaufman/subscriptions","organizations_url":"https://api.github.com/users/rgaufman/orgs","repos_url":"https://api.github.com/users/rgaufman/repos","events_url":"https://api.github.com/users/rgaufman/events{/privacy}","received_events_url":"https://api.github.com/users/rgaufman/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2018-04-18T18:33:21Z","updated_at":"2018-04-19T09:53:17Z","closed_at":null,"author_association":"NONE","body":"This is more of a question than an issue. If the device is offline for an extended period of time, is there some strategy in ruby-kafka to store the messages locally until the connection is restored and they can be sent to the broker?\r\n\r\nWhat about if the process is killed, is there any way to have a persistent store of these pending messages to be sent when the connection is restored?"},"comment":{"url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/comments/382678043","html_url":"https://github.com/zendesk/ruby-kafka/issues/568#issuecomment-382678043","issue_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/568","id":382678043,"user":{"login":"dasch","id":6351,"avatar_url":"https://avatars0.githubusercontent.com/u/6351?v=4","gravatar_id":"","url":"https://api.github.com/users/dasch","html_url":"https://github.com/dasch","followers_url":"https://api.github.com/users/dasch/followers","following_url":"https://api.github.com/users/dasch/following{/other_user}","gists_url":"https://api.github.com/users/dasch/gists{/gist_id}","starred_url":"https://api.github.com/users/dasch/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dasch/subscriptions","organizations_url":"https://api.github.com/users/dasch/orgs","repos_url":"https://api.github.com/users/dasch/repos","events_url":"https://api.github.com/users/dasch/events{/privacy}","received_events_url":"https://api.github.com/users/dasch/received_events","type":"User","site_admin":false},"created_at":"2018-04-19T09:53:17Z","updated_at":"2018-04-19T09:53:17Z","author_association":"CONTRIBUTOR","body":"ruby-kafka doesn't offer anything out of the box, but as @nguyenquangminh0711 says, you can built something yourself. For local disk persistence, this could work:\r\n\r\n```ruby\r\nrequire \"kafka\"\r\nrequire \"json\"\r\n\r\nclass PersistentProducer\r\n  BUFFER_FILE = \"kafka-message-buffer\"\r\n\r\n  def initialize(producer:, buffer: [])\r\n    @producer = producer\r\n    @buffer = buffer\r\n    @file = File.new(BUFFER_FILE, \"w\")\r\n\r\n    # Cheat and reuse the encoder from ruby-kafka.\r\n    @encoder = Kafka::Protocol::Encoder.new(@file)\r\n  end\r\n\r\n  def produce(key:, value:, topic:, partition_key:)\r\n    message = { key: key, value: value, topic: topic, partition_key: partition_key }\r\n    @buffer << message\r\n    persist(message)\r\n  end\r\n\r\n  def deliver_messages\r\n    @buffer.each do |message|\r\n      @producer.produce(**message)\r\n    end\r\n\r\n    @producer.deliver_messages\r\n  end\r\n\r\n  def self.restore_from_disk(producer:)\r\n    buffer = []\r\n\r\n    File.open(BUFFER_FILE, \"r\") do |file|\r\n      decoder = Kafka::Protocol::Decoder.new(file)\r\n\r\n      until decoder.eof?\r\n        data = decoder.bytes\r\n        message = JSON.parse(data)\r\n\r\n        buffer << message\r\n      end\r\n    end\r\n\r\n    new(producer: producer, buffer: buffer)\r\n  end\r\n\r\n  private\r\n\r\n  def persist(message)\r\n    data = JSON.dump(message)\r\n\r\n    @encoder.write_bytes(data)\r\n\r\n    # Make sure the data is written to disk.\r\n    @file.flush\r\n  end\r\nend\r\n```"}}
{"actor":{"display_login":"dasch"}}