Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Do not silently discard undelivered messages #34

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
+16 −10
Open

Do not silently discard undelivered messages #34

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
831f2b9
Do not silently discard undelivered messages
jsmucr Nov 28, 2017
1dc9a3d
Updated the file to be compatible with 6.0.0
jsmucr Nov 30, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 16 additions & 10 deletions lib/logstash/outputs/tcp.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,8 +51,6 @@ class LogStash::Outputs::Tcp < LogStash::Outputs::Base
# SSL key passphrase
config :ssl_key_passphrase, :validate => :password, :default => nil

config :message_format, :validate => :string, :obsolete => "This setting is obsolete. The event will be formatted according to the codec used"

class Client
public
def initialize(socket, logger)
Expand Down Expand Up @@ -85,8 +83,12 @@ def setup_ssl
require "openssl"

@ssl_context = OpenSSL::SSL::SSLContext.new
@ssl_context.cert = OpenSSL::X509::Certificate.new(File.read(@ssl_cert))
@ssl_context.key = OpenSSL::PKey::RSA.new(File.read(@ssl_key),@ssl_key_passphrase)
if @ssl_cert
@ssl_context.cert = OpenSSL::X509::Certificate.new(File.read(@ssl_cert))
if @ssl_key
@ssl_context.key = OpenSSL::PKey::RSA.new(File.read(@ssl_key),@ssl_key_passphrase)
end
end
if @ssl_verify
@cert_store = OpenSSL::X509::Store.new
# Load the system default certificate path to the store
Expand Down Expand Up @@ -150,15 +152,19 @@ def register
begin
client_socket = connect unless client_socket
r,w,e = IO.select([client_socket], [client_socket], [client_socket], nil)
# don't expect any reads, but a readable socket might
# mean the remote end closed, so read it and throw it away.
# we'll get an EOFError if it happens.
client_socket.sysread(16384) if r.any?
# Read everything first
while r.any? do
# don't expect any reads, but a readable socket might
# mean the remote end closed, so read it and throw it away.
# we'll get an EOFError if it happens.
client_socket.sysread(16384)
r = IO.select([client_socket])
end

# Now send the payload
client_socket.syswrite(payload) if w.any?
rescue => e
@logger.warn("tcp output exception", :host => @host, :port => @port,
@logger.warn("tcp output exception, will retry", :host => @host, :port => @port,
:exception => e, :backtrace => e.backtrace)
client_socket.close rescue nil
client_socket = nil
Expand Down Expand Up @@ -200,4 +206,4 @@ def server?
def receive(event)
@codec.encode(event)
end # def receive
end # class LogStash::Outputs::Tcp
end # class LogStash::Outputs::Tcp