Building an Order Integration Flow

[00:00:00]
>> Josh Long: So let's actually take a look at a simple example here. And I've actually got the example in the source code here. Okay, and it's a pretty trivial example, I think. I'm pretty sure I put the JSON config files. Let me see if I did. So it's called Messaging integration, and I'm going to call it Enterprise Integration Patterns, EIP.

[00:00:18]
I should have a folder also in that root folder. It's called Purchase Orders. It's just a bunch of JSON files that I had CHATGPT generate for me. Okay, there you go. So it's just a bunch of nonsense JSON files, just purchase orders. So we're going to build a system to look for those purchase orders, process them, do something with them, split them up, aggregate them, and then write them out.

[00:00:40]
Are you gonna ship this to production tomorrow? Probably not, but you'll have the foundations that you could use to do so. Okay, so let me close this out here. Goodbye, Kafka, go back to the lab again. Start that Spring IO. And we're going to be using. I guess just let's use eip, okay?

[00:00:59]
We're gonna use spring integration. We're going to use the web support and maybe we use. I don't know what we want to use. I guess that's it. Let's just use Spring integration in the web support. Hit enter, uao eip.zip. Okay, remember, the nice thing about this is no golf games required.

[00:01:20]
This will work in your machine, in your infrastructure tonight if you wanted to. You don't have to schedule a meeting. So what are we going to do again? You're defining an integration flow, right? That's the thing that arranges the sequence of steps that you're trying to achieve, okay?

[00:01:36]
And our integration flow is going to be an order integration flow. It's going to handle purchase orders, that kind of thing. Nothing particularly interesting there. Just a very generic kind of thing. The way you typically do it is you do integration flow from. Well, what's the from going to be?

[00:01:54]
So you could put the inbound adapter here directly, right? And I'm going to use files. That's gonna be my inbound domain. So you don't need to install anything in particular to make this work. I'm going to do spring integration and I'm going to add spring integration file, okay?

[00:02:13]
And again, that could add spring integration Kafka. I could add spring integration Pulsar. I could add spring integration mongodb. Or whatever, I mean, there's all sorts of modules you can add here. And then I'm going to. Let's keep it simple. We'll do the files inbound adapter. Okay, so file.inbound adapter.

[00:02:37]
And I'm going to look at a directory of purchase orders. So that directory is. Let me make my life a little bit easier here. I'm gonna cheat, copy -r -/Desktop/frontend/messaging-integration/purchase-orders, okay? -/Desktop/po, nope, that's not right. Just call it purchase orders. Okay, there we go. Okay, @Value("file"://$(HOME)/Desktop/purchase-orders, yeah? So I'm going to inject it as a file, good, file.

[00:03:20]
And I'll tell it I want to create the directory if it doesn't exist, okay? And then that's it, that's the inbed adapter, okay? And you can see the type here is a file inbound channel adapter spec. But basically, ultimately it's an adapter of some sort. Files inbound adapter.

[00:03:37]
And I'm going to handle it. I'll say new generic handler and I'm going to get a object of type file there by default. So the output, the thing that we get as a result of the inbound adapter is going to be a Java I O file object. What can I do with that?

[00:03:53]
Well, I can't. I mean, it's a file. What I want to do is actually transform it. So maybe I'll do a new file to string, right, FileToString. Yeah, transformer. Do that. Now, the inbound component type for this is a string because I've just transformed it. Right, so file string.

[00:04:11]
And now I'm dealing with a string. And I'm just. Whenever you wanna do a generic sort of intercept, if you wanna just kind of look at what's going on here, use a handle, okay? So I'm creating these components. I'm stitching them together, okay? So let's just try confirming that this has worked.

[00:04:26]
And I'm actually dealing with stuff here correctly, okay? So system out. There we go. This becomes a lambda. Of course it does. I'm going to start this up. Let's see what it gives us. It should give us a lot of messages. Unless I have typed it incorrectly. I did, awkward.

[00:04:47]
Okay, take five. And now that since I typed it incorrectly, I'll have this. Wait, do I have. This is wrong. And that's wrong. It's because I did this wrong. So here we go. Okay, good. Take five. 112, restart. Okay, and do we have data in there? We don't have data in there either.

[00:05:14]
Okay, so let me see. Copy -/Desktop/frontend-masters/messaging-integration/purchase-orders, all that goes into -/Desktop/frontend-master. No, -/Desktop/purchase-orders. Okay, so this is gonna cause my application to freak out. You see that it's going to deliver one message per second. So the question then is, how did it know when to consume the messages, right?

[00:05:43]
So file systems don't tell you when something is done. You have to ask, you have to pull. Okay, some things can tell you when something is done. So for example, POP is an email protocol. You have to pull imap. You can establish a socket and it'll push, right?

[00:06:03]
But the problem is you're keeping a network socket open. It's bad for battery. So a lot of the old blackberries and trios for example, would be like. We support both, right? If you want, if you don't mind, if you want your, if you gotta get that email right now, we can open up an imac connection for you, but warning your trio will run out of battery in 20% less time or whatever.

[00:06:21]
You know, something like that, right? So if you don't mind opening up a socket and keeping a connection, IMAP is gonna give you instantaneous notifications when there's a new email pop you have to pull. This is a discussion you have to have for every protocol with which you integrate databases.

[00:06:34]
You want to query a SQL database whenever there's a new message. Databases don't push, except for postgres, which has an extension, which is weird, but ignore that. Databases don't push. Messaging Queues do though. RabbitMQ, Pulsar, Kafka, they can all tell you when something has changed. So that's a distinction that you have to understand and care about in the inbound adapter.

[00:06:54]
In this case, the file system does not push, you have to pull it. So by default, if it requires polling, spring negation will poll P O L L, not P U L L every one second. But you can control that. You can configure the poller here. So I'm gonna say pm., let's say, every Tuesday I could do a crime expression, right?

[00:07:15]
Every Tuesday at noon I'm going to pull the messages. Or I can do a quant expression, or I can do a fixed rate dot, fixed delay or fixed rate. So every 1,000. Every 1,000 what? Well, every 1,000 durations. So actually duration of minutes there, I can now pull one every minute.

[00:07:41]
So you have full control over the frequency and the way that the polling is done. But by default it's one second. So even though the embed adapter did a sweep of that directory and it said, hey, there's a dozen files here, it knows about all dozen of them, but it's not giving you all 12 of them at the same time.

[00:07:57]
It's buffering. So it gives you one message every second because that's the puller. So it means that this is a staged event driven architecture already. The whole point of staged event driven architecture is you have a bunch of steps in a sequence. Each of those steps might get overwhelmed.

[00:08:11]
So you serialize the results that go into those things. They get absorbed by the intrinsic buffering nature of the thing in that component and then they get delivered onward. So when you do true pipes and filters, staged event driven architecture, you use something like Kafka or RabbitMQ. You send a message to Kafka, deliver it to a component, process it, deliver it to Kafka, deliver it to a component, process it, et cetera, et cetera.

[00:08:34]
Okay, so this is easy. I got the data. Yeah, not very interesting though, but it's a pretty simple example. So what I want to do now is actually turn this JSON into a Java object. And I happen to have three Java objects that I've taken the time because I care.

[00:08:50]
I've pre written these six lines of code. Now I don't want to do these lines of code live because also I care. So these are Java objects that I can use to map that JSON into a domain model. So purchase order, which in turn has a set of line items and the line items, we don't need this just yet will eventually be turned into that.

[00:09:12]
So these are the two domain models that can map our JSON. So what we're going to do is we're going to say new file to string transformer and then we'll do a transform yet again, new JSON to object transformer purchase order .class. So now when I get down to this point, the payload is a purchase order.

[00:09:32]
So if I do payload, dot, line items, foreach, et cetera, let me make sure this is a little bit longer. Okay, there you go. So you can see line item. Okay, stop, line item, sku, line item, blah, blah, blah. So now I've read data in from these files and I'm processing, I'm also printing out the headers.

[00:09:56]
Notice that each envelope, each message has a payload, which is the type that we've converted, and some headers that give you the out of band information about the payload, the origin of it. Like in the case of the file, since it's a file, you can see there's a header called file name and the original file name.

[00:10:11]
So if I wanted to print that out, I could say headers, get file, original file. Okay, system out. There you go. Let me just copy and paste it. Typing. Not my thing. Okay, there we go. So here's the original file. These are out of band information, supplementary information that are included in the body of the request.

[00:10:38]
See that headers purchase order. Okay, so I've taken the string, turned it into an object, taken the object, I've got purchase orders. What can I do with it now? Well, I mean just about anything, right? Maybe I want to split it along the lines of the line items.

[00:10:54]
So this is a payload. If you have a component here that returns null, that will stop the subsequent execution of the flow. So I'm going to use payload, I'll continue the flow and I'm going to split it. And so I have to tell it how to split. I'm going to say given a type of purchase order, I want to take that and return some data here, right?

[00:11:16]
I guess I could do line items. I mean, you could do that, right? And then now at the end of that, if you do this, if you do a generic handler here, what you've got is a line item object. And then I can print out got line item payload, good.

[00:11:37]
So run that. So there you go. You can see now I'm getting a chance to process each one of these and you can do subflows. This could be. I split it here. Now I've got a separate flow that's branching out. I can do a subflow where I actually call another integration flow where I say, okay, send it to Kafka, wait for the thing to come back and then process it, that kind of thing.

[00:11:56]
So split, and then of course, once I've handled it, I can process the data, move that back again, and then you can do aggregate. And then you should at the end of this. And the way it's doing the aggregation is of course it's got headers that tell it you this is message 1 of 2 or 1 of 5 or whatever it was when you did the split.

[00:12:17]
So system out. Final payload is this. So the final payload. Stop it, stop, there we go. Final payload is this, it's a list of line items, it's aggregated the original ones back. So you did the split. I could have transformed them along the way. I didn't in this case.

[00:12:40]
But in this case, I split the purchase order into a bunch of line items. I could have transformed the line items into a network call or a status check or something like that. Then I aggregated all the messages back into one aggregate, which gave me all the split off line items back together in one list.

[00:12:55]
So I started back where I started. So this is basic integration flow 101. There's a component here for routing, for splitting, for control bus, for all this stuff, transformations, et cetera.