Learning fp-go question

[franchb]

Hello!

I'm currently in the process of learning fp-go and am aiming to apply functional programming principles in Go. To get hands-on experience, I've taken a straightforward task: rewriting a Go function that initializes a NATS JetStream connection and subsequently creates multiple streams.

Here's the original code for context:

func Fx(lc fx.Lifecycle, natsConn *nats.Conn) FX {
	lc.Append(fx.Hook{OnStart: func(ctx context.Context) error {
		js, err := jetstream.New(natsConn)
		if err != nil {
			return err
		}
		if err = GetCreateEventsStream(ctx, js); err != nil {
			return err
		}
		if err = GetCreateAlertsStream(ctx, js); err != nil {
			return err
		}
		if err = GetCreateDLQStream(ctx, js); err != nil {
			return err
		}
		return nil
	}})
	return FX{}
}

Could someone guide me on the idiomatic way to rewrite this function using fp-go? I'm looking for a functional approach that would align with the paradigms and best practices of the library.

Thank you!

[CarstenLeue]

Hi @franchb I am not familiar with neither the fx package nor the jetstream package. From what I understand your code is doing the following:

• it makes use of the fx dependency injection mechanism
• one dependency is natsConn *nats.Conn which must have been provided elsewhere
• the function you are showing is probably one you want to provide to a fx.Provide call and it generates an FX instance (I suppose this is your own service type) based on the natsConn dependency. As part of this creation you create a new stream and three event streams
• what I cannot see in this code is the shutdown logic of these streams. I guess you'll have to close js and then also those streams that have been constructed without errors
• also since the GetCreateXXXStream methods are not returning anything, I assume the are pure side effectful functions

My first advice would be to split this one dependency into four:

• one for the js instance (with proper shutdown), dependent on natsConn
• one for the stream represented by GetCreateEventsStream, dependent on the js dependency
• one for the stream represented by GetCreateAlertsStream, dependent on the js dependency
• one for the stream represented by GetCreateDLQStream, dependent on the js dependency

Each of the GetCreateXXXStream should have their proper shutdown hook, so you might want to give GetCreateXXXStream a return value that identifies the stream, so you can shut it down

The advantage is that now it's in the responsability of the fx framework to handle the error scenarios, so you do not need it in your code and you also do not worry about ordering.

None of these aspects is directly related to functional programming though.

In terms of fp-go I suggest that the signature for GetCreateEventsStream could look like the following:

import (
	RIOE "github.com/IBM/fp-go/context/readerioeither"
	"github.com/nats-io/nats.go/jetstream"
	F "github.com/IBM/fp-go/function"
)

func GetCreateEventsStream(cfg jetstream.StreamConfig) func(js jetstream.JetStream) RIOE.ReaderIOEither[jetstream.Stream] {
	return func(js jetstream.JetStream) RIOE.ReaderIOEither[jetstream.Stream] {
		return RIOE.Eitherize1(js.CreateStream)(cfg)
	}
}

and you can call this as:

import (
	"context"

	RIOE "github.com/IBM/fp-go/context/readerioeither"
	F "github.com/IBM/fp-go/function"
	I "github.com/IBM/fp-go/identity"
	IO "github.com/IBM/fp-go/io"
	IOE "github.com/IBM/fp-go/ioeither"
	"github.com/nats-io/nats.go/jetstream"
	"go.uber.org/fx"
)

func Handler(lc fx.Lifecycle, js jetstream.JetStream) {
	name := "mystream"
	stream := F.Pipe2(
		CreateStream,
		I.Ap[func(js jetstream.JetStream) RIOE.ReaderIOEither[jetstream.Stream]](jetstream.StreamConfig{Name: name}),
		I.Ap[RIOE.ReaderIOEither[jetstream.Stream]](js),
	)

	lc.Append(fx.Hook{OnStart: func(ctx context.Context) error {
		return E.ToError(stream(ctx)())
	},
		OnStop: F.Bind2nd(js.DeleteStream, name),
	})
}

Stream creation looks a bit awkward because of the type hints we still have to give to go, it's in its canonical form, a curried function, then a sequence of Ap invocations to bind it.

You can also write

stream := CreateStream(jetstream.StreamConfig{Name: name})(js)

An alternative to OnStart is:

OnStart: func(ctx context.Context) error {
		return E.ToError(stream(ctx)())
	}

which is not as idiomatic but it looks nicer :-)

I must admit though, that the functional syntax does not look convincing for this usecase.

Let me ponder about this a bit more ...

Note: from a functional perspective the method CreateStream should return an effect, because it - inevitably - modifies the js object. The effects are push to the right as far a possible, which is why the unmodifiable cfg object is the first parameter of the curried function.

[franchb]

Much appreciated, @CarstenLeue! I plan to experiment with the sample you provided and consider sharing my insights in a blog post.