don't use an underscore in package name

don't use an underscore in package name

We only had python test..

It's really great to have these tests in.

Not sure if your editor adds udp here?

Not a fan of the naming here but as it's not user facing we can figure out the "correct" name when we have multiple use cases. This will make it easier.

More I think about it.. using source instead of inputsource would make sense. @ruflin @andrewkroh

If we call this one source, what should we call this one then? https://github.com/elastic/beats/blob/master/filebeat/harvester/source.go

I suggest to get the PR with this name in an iterate on top of it. There will be quite a few more renames I assume.

It's really great to have these tests in.

Are there any bounds on these values? If so you should add a validate tags like validate:"min:0 max:65535" or validate:"required" (double check the exact ucfg syntax)

Again, there was bounds before, but for now we can do the following: required and > 0, enforcing a top level bound might not be a good idea in a 6.x release.

I think the callback should include the source address (ip/port). This will allow for source metadata to be added to the event. Or in the future it could allow for per source buffering.

I recommend having Start() if you have a Stop() to keep the interface balanced and remove the responsibility of managing a goroutine from the user.

You could create a Logger instance for the server to use.

  log: logp.NewLogger("udp").With("address", config.Host),

Then anywhere within this object that you do logging you would use s.log.Xyz(). This gives you a named logger with context plus gives you the ability to use the structured logging methods if needed.

I think the deadline only needs to be set once after initialization.

Sadly this is not how it works, SetDeadline is an absolute time. But it does make sense, let say that you are waiting for a large computation you could give him a bigger deadline than a simple req/response.

	// A deadline is an absolute time after which I/O operations
	// fail with a timeout (see type Error) instead of
	// blocking. The deadline applies to all future and pending
	// I/O, not just the immediately following call to ReadFrom or
	// WriteTo. After a deadline has been exceeded, the connection
	// can be refreshed by setting a deadline in the future.

I was incorrectly thinking this was a read timeout with a time.Duration.

I'd prefer for Stop to block until the goroutine exits. This should Close the socket and wait for run to exit (via sync.WaitGroup).

Signaling to stop via a flag means that the socket won't actually be closed until the a packet is received to unblock the recvfrom or a timeout is reached (max 5 minutes).

Correct, you are right we need to get out of the loop and wait until the processing is done:

1. Close the Socket.
2. Set false, so we are not running (mostly used in tests)
3. Block until goroutine exits.

This logic looks incorrect. I think it should be if !ok || !e.Timeout(), but personally I would probably special case the timeout and continue for that.

if ok && e.Timeout() {
  continue
}
u.log.Errorw("Error receiving from socket", "error", err)

Yes its more clear that way.

Perhaps there should not be a default and make the user explicitly state on which port to listen. The logstash udp input does this.

I agree with that, but can only do in for 7.0 since this could break existing configuration.

Note, the TODO was there before, my changes. :)

+1 on the proposal from Andrew. Sorry about the TODO ...

Is the udp input feature listed as experimental or beta? This could give us some leeway to make the change now.

Lets do it in another PR, with a clear intend in the changelog.

Double colon in error message.

Typically a start method does not block like this does. It will "start" some kind of worker goroutine. This encapsulates the logic required to manage a goroutine which is good so that every user is not re-implementing the same thing. As a barebones example:

func (x *Thing) Start() {
    x.wg.Add(1)
    go func() {
        defer r.wg.Done()
        x.run()
    }() 
}

func (x *Thing) Stop() {
    close(x.done) 
    x.wg.Wait()
}

func (x *Thing) run() {
    for {
        select {
        case <-r.done:
            return
        default:
        }

        // Do a unit work.
    }
}

In your case the socket would replace the done channel. And Start would initialize listening socket before starting the worker goroutine. This way Start can return an error if it can't "listen" and when start returns the Server is guaranteed to be in the "listening" state (which might remove the need to have IsRunning).

Thanks for the long input, really appreciated and it clarify things.

I recommend listening only on localhost to avoid opening the port up to the world. Secondly, since this is a test that shouldn't fail if 10000 is already in use you can use localhost:0 and it will bind to an ephemeral port. You'll need to expose the local address from the server so you know what port that it is bound to.

If the port in the address parameter is empty or "0", as in "127.0.0.1:" or "[::1]:0", a port number is automatically chosen. The LocalAddr method of PacketConn can be used to discover the chosen port.

https://golang.org/pkg/net/#ListenPacket

What was the failure that led to this? I'm not a fan of doing this because it could be masking issues with the code or the test.

because of the platform differences we have to deal that here. I've decided to just use predicates to route the correct logic. If the tests become more complex it will make sense to use tags to target them.

I think we talked about this and you were going to open an issue for it. If you did/do can you add a link back to here.

There's a race condition problem with using an atomic.Bool to protect the underlying state of the server. A mutex needs to be used instead to protect it.

Correct, let me switch that to a mutex.

I will do the same change in the TCP.

I've moved it to a mutex instead and will do a followup issue, I need to get a bit more insight about was is managing the input before proposing something. I did the same changes in the TCP.

@andrewkroh ^