Joystream is an open source platform, with an open blockchain at its core, powering multiple video streaming Apps

Joystream

joystream

If you’d have googled dao only like 5 years ago, the majority of search results would have been about taoism (also known as daoism). It is an ancient philosophy and religion, that arose in the 6th century BCE in China. According to the , “The Tao can be roughly thought of as the flow of the Universe or as some essence or pattern behind the natural world that keeps the Universe balanced and ordered”. If you google it now, the results are mostly about “member-owned communities without centralized leadership”. Wikipedia Nowadays, things change very rapidly not only in Google search, which just reflects these changes. People create new forms of organization, invent new technologies and improve different spheres of life. But what we want to achieve is balance, we want to live in harmony with self, others, and nature. In the rapidly changing world it is getting more and more difficult to gain balance, and traditional organizations, which lack flexibility and can’t quickly adapt to changes, are beginning to be replaced by more flexible forms. That’s how the waterfall model was replaced by agile frameworks in software development. That’s how traditional business models are starting to be replaced by DAOs, a new experimental form of making business, which appeared due to the invention of blockchain. Balance in your life and the DAO In taoism the balance is summarised in Yin and Yang principles, which see the world with complementary forces, such as light and dark, active and inert, or hot and cold. They are everywhere in our lives. When we notice somebody goes to extremes, we want to go to the opposite extreme in an unconscious attempt to restore the balance. Here we need to remember that restoring the balance does not mean rejecting the opposite point of view, which only makes things worse. It is just adding color. If everything is getting too black, add white. If it’s all white, add black. The truth is always somewhere in between, and when you see the world in all its colors, you feel so alive and inspired… That’s what DAOs do. It’s not only about business and making money. What they try to do is to create a well-balanced organization that is like a self-evolving organism, and thus to get maximum efficiency and inspiration for all its members. In this ideal world, nothing should be imposed. Everyone should be heard, everyone’s opinion matters. Very much unlike boring centralized organizations, whose important decisions are driven by one person or a group of people, who often care only about money. DAOs still have a long way to go. Many of them are still insufficiently decentralized and have inconvenient tools for operating and decision-making. But it’s very interesting to see how they appear and evolve, the speed with which they gain devotees is impressive. My DAO experience It’s been a year since I participate in . Joystream is trying to build a decentralized video platform owned, controlled, and operated by its users. To involve people in its building and governance, it announced a Founding Members Program and is planning to allocate 20% of mainnet tokens to its participants. The program is very generous, but requires quite a lot of time and diving deep into things. For me, this experience started with pure curiosity, but gradually I became more and more devoted to the idea and philosophy of a DAO in general (I didn’t even know anything about DAOs before). Joystream DAO It’s like a job but gives you far more opportunities for self-expression and far more freedom. You can participate in working groups (devoting as much time as you want), choose Council members and even become a consul yourself, make proposals and suggest your ideas, create Bounties, and participate in them. For all these activities you get reputation and thus influence, testnet tokens, and if you achieve a Founding Member status, which usually takes several months of work, you will also get an allocation of mainnet tokens. The more you contribute, the more you’ll get! During my year’s involvement, I participated in several working groups, worked as a consul, did some translation tasks, made research and videos, and suggested different ideas. We had a lot of discussions and sometimes even quarrels, we formed political parties and debated with each other, we lost some participants during the stagnation phase and then acquired even more interesting contributors. But what was always the most important is the joy (that’s why it is Joystream, needless to say). You participate because you like it and you can always leave if you lose your joy. It’s not only about Joystream, it is about every DAO. Of course, some part of the motivation is always money, but it’s never the most important thing if you got involved in the DAO. The mission of the company, the way, and then just the process itself are becoming a real driving force. Joystream currently has a little more than 100 Founding Members, but they aim for something like 200, which still gives a lot of time and opportunities to contribute, so you can try yourself. Since the beginning, they tested two different models of the program and are still making a lot of improvements and experiments with everything. The goal is to create an effective and viable DAO by the time of mainnet, which is planned for December 2022. And in the mainnet, things are going to be even more exciting, because the budgets will explode, so a lot of money will be at stake — the total fundraise is roughly $13m at a $60m valuation. Imagine this sum will be governed entirely by token holders! Your money, your rules Joystream DAO has raised some money. It is not such a big sum compared to some other DAOs. If you open the “ ” page on Coinmarketcap, you’ll see that the biggest DAO project, Uniswap, has a 5 billion market cap. Almost the same as McDonald’s Japan, which currently has a capitalization. For May 2022, the top 5 DAO projects have already reached a 1 billion market cap. Top DAO Tokens by Market Capitalization $5.28 billion New DAOs are being formed as you read this article. A lot of money is in the hands of DAO participants. Let’s see what they’ll do with them!

DAO Philosophy and Business: My Experience With the Joystream DAO 

221 Stories To Learn About User Experience

175 Stories To Learn About Dao

Gifs have already become an integral part of our culture. Though they've been around for some time (in fact, this format was released in 1987, according to the ), they’ve become especially popular and trendy with the widespread of social media platforms and messengers in recent years. Nowadays, gifs are even studied by the scientists and are used in marketing campaigns. For example, Chanel used this image format their email marketing in 2017, as you can see below. Wikipedia But this gif is obviously the bad example - it’s boring and does not convey strong emotions, which gifs are usually associated with. The wealthy and serious audience of Chanel is not the best target for experimenting with strong emotions and jokes. In general, gifs appeal more to tech-savvy millenials, and really help brands to connect with them. That’s why these animated pictures became so popular in the crypto space, they are even converted to NFTs and are sold for big money. For example, this NFT gif of Nyan Cat for 300 ETH (nearly $474000 even in the current bear market): was sold Do you still have doubts about gifs importance? I was reading reviews of HackerNoon on Glassdoor, and accidentally saw : “Team members don’t use enough GIFs in the Slack channel, this must be improved immediately”. This was stated as the only cons of the company in September 2020, so maybe things have improved since then… this One platform to rule them all If you made a gif and want to share it with your friends, you can just upload it like you do with your images and videos, or you may choose to store it online to allow other people to use it. There are several popular platforms for publishing your gifs: Giphy, Tenor, Tumblr, Gfycat… But Tenor has one important advantage - impressive network of partners, who embeds it to their platforms. Do you know that if you publish your gif on , it will be featured on gif search everywhere: on Twitter, Discord, Telegram, Whatsapp, Viber, Facebook? Maybe also on other platforms, as I named only those which I tried myself. tenor.com Only recently did I know about it, because as a worker of marketing group of Joystream DAO I was researching competitors and stumbled upon Call to Action Initiative, organized by a Hive marketer. In posts like he encourages community members to reply on the chosen tweet for Hive promotion. Then he says that he created a gifs library on Tenor, which can be used in these replies to catch more attention. this I thought it was a good idea to create such a library for Joystream, too. There were only 2 or 3 gifs related to Joystream, if you search on Tenor or other apps using it (on Telegram, for example). So I made some gifs and started to upload them on Tenor… Is it really Google? It was a very bad user experience, with quite a lot of bugs during the process (I will not describe all of them). For one thing, there were triple notifications for every gif: “Congrats! Your gif is now being featured in joystream!”. But when I searched by the tag #joystream, I could not find any of them. Instead, there were other gifs, unrelated to the keyword. For example, gifs tagged as #stream or even #hyeongseop (a South Korean duo, it turned out). A few days later some of my gifs did appear among the first results by the tag #joystream. I shared my excitement with community members, and some of them used them on Telegram (they were easy to find by the keyword “joystream”). But after a few more days, they disappeared again. I am talking about , by the way, if you think there is someting criminal in them. these gifs I studied Google Terms of Service (yes, the platform belongs to Google) and Tenor Additional Terms. Am I violating this: “You must not use Tenor for any commercial purpose or for the benefit of any third party”? Seems unlikely, as I’m not selling anything, I’ve just added some gifs with Joystream logo. I am definitely complying with the rules. They have a lot of gifs with Youtube logo, for example. There is just something wrong with their search algorithms. Ok, the keyword “joystream” might be used by a person who is looking for a “stream of joy”, and not for the Joystream DAO. But even in this interpretation, search results are far from perfect. I also experimented with another keyword, “Gleev”, which is the name of a video platform under construction. I made a gif with Gleev logo and uploaded it to Tenor, but it’s also somewhere in the bottom of search results for #gleev, so low that I could not find it. Instead, gifs under the tag #glee are shown first. These search algorithms imperfections are strange, given the fact that Tenor is owned by Google, which is a search engine giant. Why do we need an alternative In its current state, Tenor is very inconvenient to use by crypto community - there is a good chance you won’t find what you are looking for, using its search. Because of this, a lot of gifs created by the crypto community are just getting lost after their publication on Discord or somewhere else (as a Bounty task, for example), and are never used again. With this in mind, I decided to look for Tenor alternatives on blockchain, but surprisingly, I didn’t find any. Surprisingly, because there are already so many different crypto projects, including video platforms and NFT marketplaces, but there is still no crypto “online gif search engine and database” (the definition of Tenor on ). Some ecosystem projects make it possible to create such a platform on their blockchains (Joystream is just one example), but nobody has ever tried to do it. Wikipedia If a good blockchain alternative to Tenor would be created, would Elon Musk, who has recently bought Twitter, replace Tenor by it on this platform? Well, he might, given his love of blockchain. Would other social media platforms and messengers follow? They always follow the trend, and crypto is becoming a trend indeed, so why not?

The Lord of the GIFs: Why We Need a Decentralized Tenor

or decentralized autonomous organizations have become one of the most powerful and widespread applications of distributed ledger technologies. DAOs Being an innovation based on blockchain technology, DAOs are transparent, auditable smart contracts intended for the purpose of coordinating groups of individuals, with a vested interest, in making governance decisions on a unified subject matter. Typically consisting of parameters relating to proposing, voting, and implementing structural changes to an application’s codebase or operational system, DAOs can be built around any arbitrary purpose. They are vehicles for the dissemination of power and the coordination of a distributed group of people around a common interest. DAOs redefine the model of arriving at social consensus. Traditionally group structures existed in a hierarchical “master-slave” model where a central organizer(s) was tasked with execution and demanded to be trusted. DAOs supersede the traditional model by providing three key functional benefits Displace hierarchical structures with horizontal ones. This is where the D in DAO comes in. Power over organizations is no longer centralized, rather, it is proportionately distributed amongst all participating members. This minimization of reliance on a single entity results in a more efficient flow of processes and strengthens the relationship between involved parties. It’s crucial to note that there is still some element of centralization risk present in DAOs, namely in the form of developer/deployer key controls. Equal opportunity for access By removing arbitrary hurdles of involvement present in legacy organizations ( ), DAOs create a more inclusive, equitable playing field for talent to be attracted to. While capital does play a role in power accrual, some models of governance reward activity over assets, allowing for less financially equipped people to earn their stake in an organization. no government approval or degree required Value/Capital allocation As an organization, a DAO must have something of value over which to govern. That is why, nearly every DAO will have a treasury ( wallet) that belongs to its community. With a hard-coded set of rules revolving around its utilization/management, individuals can influence how/what those funds are used for. This is evolutionary in terms of the depth of stakeholder involvement. Think about this from the standpoint of something like Apple. Shareholders may be entitled to owning a portion of the company and some of its earnings, but they have no say in how Apple uses its war chest of money. multisig *There can be arguments made from both sides whether it is good or bad, but the point here is that with a DAO it is at least possible. Different Types of DAOs: DAOs can be seen as modern-day business structures in cyberspace (the internet). Whereas in the real world, a business can be set up as an LLC or INC; in cyberspace, businesses are set up as DAOs. But the need for coordinating people spans far beyond just business. Venture DAOs Also called investment DAOs, these organizations pool capital from their members and act as pseudo-asset managers. Intuitively perfect for this function due to the inherent transparency of blockchain, as of April 2023, Venture DAOs alone are holding north of $3 Billion USD in their treasuries (not including the value of deployed capital). Examples of Investment DAOs: BitDAO Protocol DAOs This is the most widespread model for DAO governance. Used by nearly every relevant DEFI project, protocol DAOs exist as the backbone systems of decentralizing the ownership of software applications. Examples of Protocol DAOs: Compound, UniSwap, MakerDAO Service DAOs A mixture of talent agencies and freelance platforms, service DAOs exist for the purpose of finding providers skilled in their crafts and then connecting them with those in need of such services. Think of service DAOs as job marketplaces. Although, services can extend to unique on-chain functions such as smart contract audits. Examples of Service DAOs: TalentDAO, RooTroop, WGMI, DeveloperDAO, MetaVerse Collector DAOs Arriving on the scene after the rise of NFTs, collector DAOs are exactly what they sound like. Organizations that aggregate funds from their community members to purchase valuable on-chain objects. Those objects are then deposited into a vault owned by the DAO and in turn, provide exposure to its members. Examples of Collector DAOs: SharkDAO, Pleasr Grant DAOs GrantDAOs are for virtuous empowerment. Funds are drawn together and community members allocate the money for causes relating to helping bootstrap promising projects. This is a nuanced model that accrues its own value vicariously by carefully fostering innovation towards protocols that it is itself involved in. , Examples of Grant DAOs: GitCoin MetaCartel Media/Entertainment DAOs Everything from gaming and video streaming to art IP and advertisement networks this structure of DAOs is geared for creativity and content creation. , Examples of Media DAOs: , Bankless RugDAO, JoySteam StudioDao Social DAOs Popular model of organization for minimally financial causes, social DAOs are focused on the inclusivity of as wide an audience as possible, basically a glorified social media platform. Building one can be made possible through the use of traditional Web 2 social platforms and a multi-sig wallet with a few ETH. Examples of Social DAOs: Friends With Benefits DAO OS — Operating Systems Operating systems are purpose-built protocols that enable other organizations with tooling to deploy their own DAOs. In order to uphold their integrity, these operating systems are also structured and run as DAOs. , Examples of DAO Operating Systems: Aragon, Colony, DAOstack, DAOhause Tally, XDAO DAO Governance Models: Whenever a DAO is set up there must be some way to specify membership and inclusion; this is generally done through one of two ways: tokenization and/or contribution. At the heart of a DAO exists its token. Commonly associated as being units of account and instruments of value/influence; in the case of DAOs, tokens represent proportional shares of the right to participate in voting. However, organizations are well aware of the possible capital constraints that can exclude promising members from joining on the merits of being able to acquire tokens alone. Therefore, incentive system rewards for active participation have become a popular tool for token emission, allowing people without money a chance to earn their way into a DAO through their efforts. Once tokens are acquired and users can participate in a DAO, there are three general models for their involvement: Automatic/Direct This is the most desirable form of governance as it involves all users independently, directly participating in all activities. Once a consensus threshold level is reached, the smart contract will automatically execute the specifications without the need for a single authority to intervene. Delegated A model that has long been utilized for consensus mechanisms in distributed systems, delegation is something vaguely similar to representative democracy. Here, community members offload their responsibilities to other community members and trust them to make decisions on their behalf. Important note, tokens are not given to other members, only their weight equivalent in influence. Multisig/Treasury A model that slightly goes outside of the natural execution automata of smart contracts, multisig governance refers to the allocation of funds into the hands of the most financially apt and trusted participants of the community that reveal their identities publicly. So even if there are 10,000 voters passing a vote, they will not be able to influence the treasury without the involvement of the multisig entities. The main reasoning behind this formulation is to deter malicious anonymous entities from subverting the system for financial gains and the minimization of inadequate voting on behalf of the community. (If a DAO control $100 million USD and some proposal garners 6,000 of 10,000 votes to spend $90 million USD on minting random NFTs, it would be best if a smaller patch of clear-minded participants never executes on that.) Legal Implications of DAOs Being products of geo-neutral software systems, many assume that DAOs are not subject to the same regulatory frameworks as traditional businesses. That has not been completely the case. Government organizations, namely the CFTC, have actually offloaded the burden of legal implications onto the shoulders of the token holders and the developers possessing the private keys for conducting upgrades ). (for example with bZx & Ooki It is still unclear as to how things will pan out, but as it stands, decentralized autonomous organizations are slowly materializing into hybrid traditional organizations. Some authorities around the world have begun to acknowledge DAOs as real-world entities by providing legal frameworks to recognize DAOs as equivalent to LLCs. In any case, the future will be digitized, we can sit in uncertainty trying to figure out how the legal system may integrate with DAOs, or we can understand these systems better and do something awesome with them and then have governments catch up to us retroactively. Let’s opt for the latter. Closing thoughts From the very first iteration of the “The DAO” which raised >$150 million USD and caused the infamous hard fork split of Ethereum in 2016, to today's recognition by nation-states as legitimate entities, the landscape of DAO is quickly evolving. AI technologies have captured the fantasies of mankind, and their applications have found direct, meaningful ways to integrate with blockchain. Delphi Labs has put out a whitepaper in which it sees the synthesis of ) to advance upon the shortcoming in the portion of existing DAOs by introducing artificial agents. DAOs with AI (called BORGs Autonomy There is no doubt in my mind that DAOs as a sector/metric for the representation of growth in the digital economy will continue to flourish and empower people around the world with the freedom to pursue radical new ideas. Stay interested. Stay involved. There is a renaissance taking place, and everybody who can rise above the noise of negativity is invited to participate. We are no longer at the beginning, but damn, we are still so early. See you all on the other side 🥂 Also published . here

How DAOs are Transforming Social Coordination and Governance

About a year ago, I was working on a project at my previous job. The goal was to set up a reliable message queue. Most people might go straight to RabbitMQ, and I almost did too. But I decided to look for something different. That's when I came across NATS JetStream. After working with it for the past year, I can say it's truly impressive. It's fast, it can handle a lot of data, and it fits well with cloud systems. Today, I'll walk you through how to use NATS JetStream with a Golang app. By the time you finish reading this article, you'll be able to implement a robust and fast message queue. Message queues For those who are not familiar, what are message queues? In the real world, message queues help systems talk to each other without getting overwhelmed. For example, when you order something online, your order might go into a message queue before it's processed. This way, even if many people order at the same time, the system can handle it smoothly. Message queues are key for making sure data gets where it needs to go, especially when things get busy. Project requirements For my project, I had a clear list of needs when it came to selecting a message queue: : It was crucial that our messages stayed safe. We couldn't afford to lose them due to network glitches or if a server needed to restart. In simple terms, once a message is in the queue, it should stay there until it's processed. Durable Queues : Sometimes, there can be small issues or temporary errors when a system tries to read a message. I needed a way to send these messages back into the queue so they could be tried again later. This ensures that no message gets left behind because of short-term problems. Retrying Messages : As our system might get more users and data, I wanted our message queue to grow easily with us. Horizontal Scaling With these requirements in mind, I began my search for the perfect message queue solution. Possible solutions When searching for the right message queue, I explored several options. Here's a brief overview: : I didn't dive too deep into these. They're older and seem to lack current support. Gearman, Beanstalkd, and similar : It's favored by many small businesses. However, I found it a bit slow. Setting the HA (High-available) cluster requires real expertise. Plus, it lacks the ability to “requeue” a message with delay. RabbitMQ : It's excellent for PubSub and storing KV data. But as a queue system? Not the best. If Redis is set to store all messages, it slows down drastically. Redis : It's a strong choice for event streaming. But it’s not the best choice for message queues. For instance, how can I requeue a message? Apache Kafka : It looked promising and fits the needs of large organizations. It met most of my criteria. Yet, horizontal scaling isn't straightforward. It demands a large infrastructure and a skilled team, which felt excessive for my project's scale. Apache Pulsar Finally, I stumbled upon NATS JetStream. It seemed to align perfectly with my requirements. How NATS JetStream fits to requirements To understand how NATS JetStream works, I made some simple code and did tests. Here's what I learned: : : Setting Things Up I made two simple Golang CLI tools One for sending messages (NATS publisher) ...
publishedTotal := 0
for i := 0; i < 30000; i++ {
	subject := "test_subject_name"
	_, err = jetStream.PublishAsync(subject, []byte(strconv.Itoa(i)))
	if err != nil {
		return fmt.Errorf("publish: %w", err)
	}

	publishedTotal++
}
fmt.Printf("published %d messages\n", publishedTotal)
... One for fetching messages (NATS consumer) ...
totalReceived := int32(0)
for {
	func() {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()

		msgs, fetchErr := jetStreamSubscriber.Fetch(1, nats.Context(ctx))
		if fetchErr != nil {
			if errors.Is(fetchErr, context.DeadlineExceeded) {
				return
			}

			log.Printf("sub fetch err: %s\n", fetchErr)
			return
		}

		for _, msg := range msgs {
			ackErr := msg.Ack()
			if ackErr != nil {
				log.Printf("sub ack err: %s\n", ackErr)
			}
			totalReceived++
			log.Printf("total received: %d\n", totalReceived)
		}
	}()
}
... Durable queues With these tools ready, I checked a few things: : Sending Many Messages : What if I send thousands of messages all at once? Question : The consumers took all the messages in. Nothing was lost. Answer : Sending Without Consumer : What if I send messages but there's no one to consume them? Question : When a consumer is ready, it receives all the messages. Answer : NATS Server Issues : How do consumers act if the NATS server stops? Do they try to connect again when NATS is back? Question : Yes, consumers connect back to NATS by themselves when it's working again. Answer : Unread Messages and NATS Problems : What about messages sent to NATS but not consumed, and then NATS server turns off? Question : If NATS starts working again, all the waiting messages are consumed. This is because NATS keeps messages safe on the computer's storage. Answer NATS showed excellence aligning our first requirement. Let’s check out how it can send messages back to a queue. Requeue In NATS JetStream, requeuing a message that has encountered a problem is straightforward. You can achieve this by calling the function available in the message API. Invoking ensures that the message is sent back to the queue, effectively providing a second chance for it to be processed at a later time. Nak() Nak() Another feature that adds to the utility of NATS JetStream is the ability to requeue messages with a custom delay. This functionality comes in handy when dealing with tasks that are deferred and need to be carried out after a certain period has elapsed. For example, if you have a notification system that must send alerts at specific intervals, you can use this feature to schedule these messages accurately. Two requirements are nailed. There is one more important point about scalability. Horizontal Scaling When it comes to horizontal scaling, NATS JetStream makes the process relatively simple. Setting up a cluster involves running multiple NATS servers with JetStream enabled. All that's required is to specify the same and define the routing paths between each node in the configuration. cluster_name I added an example of docker-compose.yml file to run a local NATS cluster: version: "3.5"
services:
  nats1:
    container_name: nats1
    image: nats
    entrypoint: /nats-server
    command: --server_name N1 --cluster_name JSC --js --sd /data --cluster nats://0.0.0.0:4245 --routes nats://nats2:4245,nats://nats3:4245 -p 4222
    networks:
      - nats
    ports:
      - 4222:4222

  nats2:
    container_name: nats2
    image: nats
    entrypoint: /nats-server
    command: --server_name N2 --cluster_name JSC --js --sd /data --cluster nats://0.0.0.0:4245 --routes nats://nats1:4245,nats://nats3:4245 -p 4222
    networks:
      - nats
    ports:
      - 4223:4222

  nats3:
    container_name: nats3
    image: nats
    entrypoint: /nats-server
    command: --server_name N3 --cluster_name JSC --js --sd /data --cluster nats://0.0.0.0:4245 --routes nats://nats1:4245,nats://nats2:4245 -p 4222
    networks:
      - nats
    ports:
      - 4224:4222

networks:
  nats:
    name: nats What adds to the utility is that you can connect to any node within the cluster, whether from a publisher or a consumer application, and it just works seamlessly. This implies that the system is designed to handle varying load distributions, routing the messages effectively irrespective of which node you are connected to. It was clear to me that NATS JetStream aligned with all my requirements perfectly. Pros and nuances I spent considerable time with NATS JetStream and can outline its strengths and limitations based on my experience: Pros : NATS JetStream is more than just a message queue system. It offers a variety of features that I'll elaborate in the next articles. Broad Functionality : Being written in Go, the source code is accessible and provides insight into undocumented functionalities. This is particularly useful for Go developers who want to delve deeper into the system. Go-Based : The performance metrics for NATS JetStream are impressive, making it a reliable choice for high-throughput scenarios. Strong Benchmarks Nuances : One thing to be aware of is that NATS ensures 'at-least-once' message delivery. As a result, your consumer logic must be idempotent to handle such cases. 'At-Least-Once' Delivery : While the documentation is generally good, it doesn't cover all configuration options. Sometimes, you'll need to go through the SDK code to understand certain functionalities fully. Incomplete Documentation : This could be seen as both a pro and a nuance. The SDK handles reconnections automatically, which means you don't have to code this logic yourself. Built-In Reconnection Logic Performance The performance benchmark was conducted by , where NATS JetStream was compared with other popular message queues like Apache Pulsar and RabbitMQ. It's important to approach these results with a degree of skepticism, as the company is known for its affinity towards Apache Pulsar. They themselves noted that they faced difficulties in configuring NATS JetStream for the benchmark, which could potentially skew the results in favor of their preferred system. Steamnative NATS JetStream – During our tests, we attempted to follow the recommended practices for deliverGroups and deliverSubjects in NATS, but encountered difficulties. Our NATS subscriptions failed to act in a shared mode and instead exhibited a fan-out behavior, resulting in a significant read amplification of 16 times. This likely significantly impacted the overall publisher performance. Conclusion After working with NATS JetStream for an extended period, several aspects of the system became apparent: Easy configuration. High performance. Broad functionality. Beyond its capabilities as a message queue, NATS JetStream also functions as a message broker for event-streaming, PubSub, and synchronous Request-Reply message bus. While this article provides an initial overview, I plan to discuss the diverse features of NATS JetStream in more detail in future articles. This will include a closer look at its specific functionalities and how they can be applied in different scenarios. References NATS JetStream Publisher/Subscriber demo code Comparison of NATS, NATS Streaming and NATS JetStream Implementing NATS JetStream as Queues in Golang Steamnative performance benchmark Steamnative performance benchmark (detailed)

NATS JetStream — a New Way to Create Resilient Message Queue

is an excellent key-value distributed database used internally by Kubernetes and managed by the CNCF. It's a great option, and that's the reason why Apache APISIX uses it too. Yet, it's not devoid of issues. etcd First, some mention scalability, but one can expect this from a distributed data store that values consistency. Another issue may be the need for more familiarity with etcd. It's relatively new, so your Ops team may need help operating it correctly while having decades of operating MySQL or Postgres. Finally, only a few etcd users are aware that it lacks maintainers: In the last few months, primary maintainers G.L. (Amazon, announcement) and S.B. (Red Hat) have stopped actively participating in the project. This leaves the project with only one active and two occasionally-reviewing maintainers, M.S. (Google),  P.T. (Google), both are relatively new to the project (1 month and 1 year of tenure) and S.P.Z. (IBM). Other maintainers are either dormant or have very minimal activity over the last six months. . The project is effectively unmaintained (emphasis mine) -- Google Groups of Kubernetes Steering Committee, March 2022 For all those reasons, you may prefer to use a standard SQL database with Apache APISIX. In this post, I'll show how you can use MySQL. The kine project It would be a lot of effort if each product had to introduce an abstraction layer and different adapters for both etcd and other databases. kine is a project that aims to offer a translation step between etcd calls and other implementations: Kine is an etcdshim that translates etcd API to: SQLite Postgres MySQL NATS Jetstream Features Can be ran standalone so any k8s (not just K3s) can use Kine Implements a subset of etcdAPI (not usable at all for general purpose etcd) Translates etcdTX calls into the desired API (Create, Update, Delete) -- Kine (Kine is not etcd) In essence, kine is a Go library that translates etcd calls to the datastore you want (among those implemented). Yet, using kine directly is a non-trivial effort. Fortunately, , the company that gave Apache APISIX to the Apache Software Foundation, provides a component already focused on APISIX usage. api7 ETCD adapter ETCD adapter wraps kine to be APISIX-specific: ETCD Adapter mimics the ETCD V3 APIs best effort. It incorporates the kine as the Server side implementation, and it develops a totally in-memory watchable backend. Not all features in ETCD V3 APIs supported, this is designed for Apache APISIX, so it's inherently not a generic solution. -- ETCD adapter Two things of note: At the moment of this writing, the adapter supports either local in-memory storage or MySQL It's available as an embeddable library but also as a standalone component Therefore, we can design our architecture as the following: Demo Let's implement the above architecture with an additional admin UI over MySQL. I'll use Docker Compose: version: "3"

services:
  apisix:
    image: apache/apisix:3.4.0-debian                         #1
    volumes:
      - ./config.yaml:/usr/local/apisix/conf/config.yaml:ro
    ports:
      - "9080:9080"
      - "9180:9180"
    depends_on:
      - etcd-adapter
    restart: always                                           #2
  etcd-adapter:
    build: ./etcd-adapter                                     #3
    volumes:
      - ./adapter.yml:/etcd-adapter/conf/config.yaml:ro       #4
    depends_on:
      - mysql
    restart: always                                           #2
  mysql:
    image: bitnami/mysql:8.0                                  #5
    ports:
      - "3306:3306"
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_USER: etcd
      MYSQL_PASSWORD: etcd
      MYSQL_DATABASE: apisix
  adminer:
    image: adminer:standalone                                 #6
    ports:
      - "8080:8080"
    environment:
      ADMINER_DEFAULT_SERVER: mysql
    depends_on:
      - mysql Latest version of Apache APISIX, yeah! To avoid any failure with dependencies between containers, restart until it works. Kubernetes's manifests would involve health checks still needs to provide a container. We need to build from the source code api7.ai Override the default configuration file with a context-specific one The regular MySQL image didn't work for me. Let's take the one from Bitnami Adminer, formerly known as PHP myAdmin, will help to visualize the database state ETCD-adapter's configuration looks like this: server:
  host: 0.0.0.0                 #1
  port: 12379

log:
  level: info

datasource:
  type: mysql                   #2
  mysql:
    host: mysql                 #3
    port: 3306                  #3
    username: etcd              #3
    password: etcd              #3
    database: apisix Bind any IP since Docker will assign a random one Implementation type. The default is ; we need to change it. btree As configured in the file docker-compose.yml Finally, here's Apache APISIX configuration: deployment:
  admin:
    allow_admin:
      - 0.0.0.0/0
  etcd:
    host:
      - "http://etcd-adapter:12379"   #1 Use this etcd instance, which is the adapter Testing Now that we are set let's test our system by creating a route: curl -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d '{                                          
  "methods": ["GET"],
  "uris": ["/get"],
  "upstream": {
    "nodes": {
      "httpbin.org:80": 1
    }
  }
}' http://localhost:9180/apisix/admin/routes/1 We can now get it: curl -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' http://localhost:9180/apisix/admin/routes/1 We can also check via the Adminer interface that it has been persisted via MySQL: Unfortunately, we need to stop at this point. Getting all routes doesn't work: curl -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' http://localhost:9180/apisix/admin/routes {"header":{"revision":"1689689596"},"message":"Key not found"} Worse, using the route fails: curl localhost:9080/get {"error_msg":"404 Route Not Found"} Conclusion etcd is an excellent piece of infrastructure Kubernetes uses, but there might be better choices in some contexts. Worse, it might become a security threat in the future - or is already one, because of the lack of maintenance. Being able to move away from etcd is a considerable benefit. kine offers an etcd-compatible facade and multiple implementations. Using kine with Apache APISIX requires some adaptation effort, already done in ETCD-Adapter. Currently, ETCD-Adapter is not feature-complete (to say the least) and requires more love. That's why it was not donated to the Apache Foundation yet. If you're a Go developer and are interested in the project, feel free to subscribe to the to offer your help. Apache APISIX mailing list and/or join our Slack The complete source code for this post can be found on . GitHub To go further: Kine ETCD Adapter "Goodbye etcd, Hello PostgreSQL: Running Kubernetes with an SQL Database" Originally published at on July 23rd, 2023 A Java Geek

Apache APISIX Without etcd?

Times are changing faster than we ever thought possible. It seems like every morning I wake up and there is a brand new revolutionary technology or political movement creating massive ripples in the world. As , and as leaders in general, we need to take the future very seriously. We live in a world that's changing at breakneck speed, and leadership has never been more crucial. entrepreneurs In business, things don’t stop moving, so if we can’t keep up and adapt on the fly, then our business is doomed. So as we gear up to dominate the next decade, we need to be aware of the challenges that lie ahead for leaders. It's time to strap on our boots and prepare ourselves to tackle the obstacles that are coming our way. , an HBS professor, I picked his brain about leadership. In my recent conversation with Bill George He has studied it his entire life and gave me a lot of insight into what it takes to make a great leader. One of the most interesting ideas we touched on that inspired me to write this week's newsletter, was the idea of the challenges leaders are going to face in the future. The world is changing at an unprecedented rate, and as a leader, it’s vital to keep a pulse on things that could be coming down the pipeline. So in this newsletter, I'm going to break down the most pressing challenges that leaders will face in the next ten years. From technological disruption to shifting demographics, we'll cover everything you need to know to stay ahead of the game. Buckle up, because it's time to get real about the future. Study The Past To Predict The Future As the futurist Alvin Toffler once proclaimed, "You've got to think about 'big things' while you're doing small things, so that all the small things go in the right direction." One way we can incorporate this wisdom is through a deep, immersive study of the past to make informed predictions about the future. Amid the dizzying speed of technological innovation, shifting market landscapes, and ever-evolving consumer preferences, it becomes crucial to take a step back and look to history. In a world that often seems besotted by the 'new,' remember that history often rhymes if it doesn't repeat itself. In fact, there are two salient reasons to for a better grip on the future: detecting patterns and trends, and gaining comprehensive contextual understanding. delve into the annals of history Unearthing Patterns and Trends Historical analysis serves as a treasure trove of patterns and trends that can provide us with invaluable insights. A systematic review of past events and data can arm us with the understanding of what strategies and approaches were effective and those that led to fiascos. Take, for instance, the infamous s. During this period, an insatiable investment frenzy in Internet-centric businesses led to a market situation where the emphasis on actual profitability was lost in the dust. Many of these businesses folded, leading to enormous financial losses. An analysis of this scenario reveals the perils of speculation, unsustainable growth expectations, and poor due diligence. Such knowledge is invaluable in spotting similar warning signs in emerging markets and enabling more prudent investment strategies. dot-com bubble of the late 1990 Gaining Holistic Contextual Understanding A secondary, yet equally important, benefit of historical study is contextual understanding. When we dissect past events and situations from different perspectives, we stand to gain a more nuanced view of the conditions that led to specific outcomes. By understanding the contextual intricacies, we can anticipate how a similar confluence of factors may precipitate comparable outcomes in the future. Take the example of the . A calamitous period in global economic history, the Depression was the result of a potent cocktail of factors: unfettered stock market speculation, widespread bank failures, and insufficient governmental checks and balances. Recognizing these as catalysts of a cataclysmic economic downturn, we can stay vigilant against policies that could potentially create a similar risk scenario. Great Depression of the 1930s The past serves as a powerful compass for the future. Its stories, lessons, and patterns help us navigate the complex landscape of the present and prepare us for the future. As Mark Twain said, "History doesn't repeat itself, but it does rhyme." By studying these historical rhymes, we can make more informed decisions and create a future that learns from the past rather than repeating it." What To Look Out For In The Future As we've navigated the lessons from the past, it's time to cast our gaze towards the future, specifically in the realms of business and leadership. The coming decade holds a multitude of challenges, and success will hinge on the ability to adapt to monumental shifts. Here are key trends you should watch as we navigate the tumultuous tides of change: Technological Disruption As we stand on the brink of a technological revolution, we face a future that's hurtling towards us at breakneck speed. Advancements in artificial intelligence, machine learning, and other emerging technologies are not merely twinkles on the horizon; they're meteor showers poised to reshape the landscape of business and leadership. They're driving us into a future where the only option is to adapt, innovate, and conquer. Navigating this labyrinth may seem daunting, but let's remember that within these challenges lie immense opportunities - golden threads woven into the very fabric of this disruption. It's about breaking free from the anchor of complacency and setting sail on the voyage of continuous learning, evolution, and growth. It's about turning these winds of change into the jetstream that propels us forward. Here's your compass: Fully immerse yourself in the waves of emerging technologies. Make it your business to understand these dynamic trends, not just as abstract concepts but as tangible tools of transformation. Dedicate time for learning sessions, tech-trend roundtables, and meaningful discussions with tech experts. As the adage goes, knowledge is power, and in this case, it's the power to propel us into the future. Dive into the Digital Deep-End: In the era of rapid technological change, the traditional playbooks of business strategy need to be rewritten. Instead of a culture that tiptoes around change, we need one that somersaults into it. Nurture an environment where teams are inspired to take calculated risks, experiment with new ideas, and see the potential of technology as a catapult rather than a cliff. Cultivate a Culture of Innovation: As the tech landscape evolves, so too must our skill sets. Partner with academic and professional institutions to develop personalized learning paths that cultivate the necessary skills within your teams. This not just equips them to ride the tech wave but also empowers them to be the wave-makers of the future. Equip Your Team for Tomorrow: Shifting Demographics With the turn of the century, we've seen the demographics of our world shift like pieces on a grand, global chessboard. An aging population in some regions, a youth surge in others, and cultural and ethnic diversities blossoming everywhere - these are not mere statistical changes. They are evolving narratives that hold the power to shape the marketplace of the future. As leaders, we need to take the pulse of these changes, understand the new rhythms, and adapt our strategies to the beats of these evolving demographics. It's about striking the right chord with our audience, not just echoing their sentiments but also resonating with their unarticulated needs and desires. Here's how we can orchestrate this: Stay acutely aware of the shifting demographic patterns, be it the changing age dynamics or cultural shifts. Listen to their stories, understand their unique needs, and tailor your products or services to fit snugly into their narratives. It's not just about market satisfaction; it's about crafting a customer experience that feels personal and purposeful. Get a Pulse on Your Audience: The diversifying cultural fabric of our world presents a beautiful mosaic of opportunities. Learn the intricate patterns, appreciate the uniqueness of each culture, and weave these insights into your organizational strategy. Fostering a diverse and inclusive team can be your guide here, enabling you to connect with a wider audience and open the floodgates of untapped markets. Embrace Globalization: Demographics are dynamic; they're the tides that shape the shorelines of our markets. Stay flexible and responsive to these changes. From altering your business model to customizing your strategies, keep your hands on the wheel of adaptability. In the end, the key to thriving in a shifting demographic landscape is not just about riding the wave, but learning to steer and shape it in your favor. Stay Agile: Climate Change As the sun rises on the 21st century, we find ourselves in an era where the call to action on climate change has never been louder or more urgent. The footprint of our businesses on this planet is casting a longer shadow than ever, making it imperative to step into the light of sustainability. What was once a moral and ethical responsibility has now become a critical business imperative. Ignoring it would be like turning a blind eye to the rising tide. Embracing it, on the other hand, opens doors to a new realm of possibilities - cleaner operations, sustainable solutions, and a future that is not just prosperous but also habitable. It's about steering our businesses from a short-term, profit-focused perspective to a long-term, sustainability-driven one. The green revolution is not coming; it's here. And here's how we can take the reins: Embed sustainability into the DNA of your business. From opting for renewable energy sources to reducing waste, there are myriad ways to transition into a greener business model. This won't just help shrink your carbon footprint, but also sprout opportunities for cost reduction and efficiency. Think Green, Act Green: As consumers become more conscious about their ecological footprint, businesses must meet them on this green pathway. Strive to design products and services that are not just high on quality but also low on environmental impact. It's not just about delivering to market demand; it's about paving the way for a sustainable future. Meet the Eco-Consumer: In the era of green innovation, keeping your eyes and ears open to the ground can give you a head start. Be it cleaner technology, sustainable supply chain practices, or eco-friendly packaging solutions, being informed can be the first step towards being innovative. Stay One Step Ahead: Globalization In the grand theater of business, the curtains have lifted on a new act - globalization. Our world has shrunk into a global village where businesses can traverse continents at the speed of a click. This new act, however, comes with its own set of rules and cues. Crossing borders is not just about geographical distances but also cultural gaps. It's about understanding foreign landscapes, respecting their nuances, and blending our strategies to fit these new molds. But let's remember that within these challenges lie untapped markets, potential partnerships, and a world of opportunities. It's about viewing the world through a wide-angle lens and turning the globe into your playground. Here's your playbook: The first step to playing the global game is understanding its rules. This involves getting a firm grasp on the laws and regulations of the international markets you wish to enter. Collaborating with local partners or hiring experts can serve as your guiding beacon, helping you navigate the complex maze of global business. Get Acquainted with the Rules: As businesses expand across borders, they must also expand their cultural understanding. This means cultivating cultural sensitivity, promoting clear and open communication, and celebrating the diversity that is intrinsic to the global marketplace. It's about building teams that are not just fluent in languages but also in cultures. Bridge the Cultural Divide: New markets are like uncharted territories waiting to be explored. Be proactive in identifying these opportunities, build robust networks with local businesses, and don't hesitate to venture into unfamiliar terrains. Globalization is not just about surviving in the global market; it's about flourishing and growing. Seize the Opportunity: Talent Management Talent management in today's rapidly evolving business landscape isn't just a strategic element - it is the cornerstone of organizational success. In an age where competition for top talent is fierce and retaining them is a Herculean task, it's about time we flip the script. Today's workforce isn't just hunting for good paychecks, they are yearning for purpose, growth, and recognition. It's about leading your company towards a horizon where talent is not just managed but celebrated, engaged, and inspired. Here's your map to the talent management territory: Creating a vibrant, inclusive culture is a surefire way to attract top talent. By fostering an environment that values creativity and innovation, you pave the way for a workforce that is not just productive but also dedicated. It's about building a workplace that is a melting pot of ideas, a platform for growth, and a harbinger of positivity. Craft a Compelling Culture: Employee engagement is the secret sauce that retains top talent. Encourage open dialogue, provide ample opportunities for skill development, and ensure a healthy work-life balance. It's about helping your employees feel a sense of belonging, a sense of achievement, and a sense of purpose. Keep the Engagement Meter High: Position your organization as an employer of choice. A strong employer brand that aligns with your company's core values can be a powerful talent attractor. It's about amplifying your strengths, sharing your success stories, and showcasing what sets you apart from the rest. Be a Talent Magnet: Cybersecurity In the digital realm where businesses are bolstered by technology, the dark shadow of cybersecurity threats looms large. Cybersecurity is no longer a mere item on the IT checklist - it is the very armor that can shield your business from dire consequences. In this era of frequent data breaches and sophisticated cyberattacks, keeping your cyber-guard up is not an option; it is a mandate. But let's not just view cybersecurity as a defense strategy; let's consider it as a robust foundation upon which a business can safely innovate, evolve, and thrive. Here's how to gear up: Your digital assets are the treasure troves that cybercriminals are after. Ensure they are fortified with state-of-the-art cybersecurity measures. From firewalls and encryption to timely security updates, make sure your digital fortress is impregnable. Safeguard Your Digital Fortress: Having a comprehensive incident response plan in place is a business imperative. From identifying the breach to containing the damage, your plan should be a well-choreographed sequence of actions that can minimize the impact of a cyberattack. Be Battle-Ready: Your employees are your first line of defense against cyber threats. Equip them with the knowledge and tools to recognize potential threats and respond promptly. After all, cybersecurity is not a one-time project; it's a constant endeavor. Empower Your Human Firewall: Wrap-Up The challenges that leaders will face in the next ten years are complex and diverse. To succeed in this ever-changing world, entrepreneurs must be nimble, flexible, and ready to adapt at the drop of a hat. But, with proper preparation and forward thinking, not only can you be ready to survive in the future, you can be ready to thrive in it. If you want to hear more about this, . He has a lifetime of knowledge and insights, and this conversation is one of my all-time favorites! be sure to check out the full conversation with Bill That’s it for now; thanks for reading. If you enjoyed this article, I’d love to hear from you. Reply to this email or at me and I'll do my best to get back to everyone! tweet @ScottDClary The lead image for this article was generated by HackerNoon's via the prompt "leaders on a mountain" AI Image Generator

The Biggest Challenges Leaders Are Going to Face in The Next 10 Years

পরবর্তী 10 বছরে নেতারা সবচেয়ে বড় চ্যালেঞ্জের মুখোমুখি হচ্ছেন

Die größten Herausforderungen, denen sich Führungskräfte in den nächsten 10 Jahren stellen müssen

Los mayores desafíos que enfrentarán los líderes en los próximos 10 años

Les plus grands défis auxquels les dirigeants seront confrontés au cours des 10 prochaines années

अगले 10 वर्षों में नेताओं को सबसे बड़ी चुनौतियों का सामना करना पड़ रहा है

リーダーが今後 10 年間に直面する最大の課題

향후 10년 동안 리더들이 직면하게 될 가장 큰 과제

Os maiores desafios que os líderes enfrentarão nos próximos 10 anos

Liderlerin Gelecek 10 Yılda Karşılaşacağı En Büyük Zorluklar

Những thách thức lớn nhất mà các nhà lãnh đạo sẽ phải đối mặt trong 10 năm tới

领导者在未来 10 年将面临的最大挑战

Building 'The Builder' - The Road from Commit to Production in 13s

Streaming data from a MySQL database to a PostgreSQL database can be a useful way to move data between systems or to create a real-time replica of a database for reporting and analysis. One way to accomplish this is through the use of tools. Change Data Capture (CDC) CDC is a method of tracking changes made to a database and capturing them in a separate stream. This stream can then be used to replicate the changes to another database. DBConvert Streams helps to replicate your MySQL data to PostgreSQL in real-time. It captures data changes from a source MySQL database and applies them to a target PostgreSQL database. This can be done by setting up a source to read the binary log of a MySQL database and transform the changes to a format that can be consumed by target PostgreSQL database. [Database Streaming Examples and configurations GitHub repository](https://github.com/slotix/dbconvert-streams-public) To begin, let's clone the GitHub repository containing the MySQL to PostgreSQL streaming example. git clone git@github.com:slotix/dbconvert-streams-public.git && cd dbconvert-streams-public/examples/mysql2postgres/sales-db/ Docker Compose configuration. Since DBConvert Streams relies on multiple services, the most efficient way to start the containers is by using . Docker Compose The file from the repository is provided below. docker-compose.yml version: '3.9'
services:
  dbs-api:
    container_name: api
    image: slotix/dbs-api
    entrypoint:
      - ./dbs-api
      - --nats=nats:4222
      - --source=source-reader:8021
      - --target=target-writer:8022
    ports:
      - 8020:8020
    depends_on:
      - nats
    volumes:
      - ./mysql2pg.json:/mysql2pg.json:ro

  dbs-source-reader:
    container_name: source-reader
    image: slotix/dbs-source-reader
    entrypoint:
      - ./dbs-source-reader
      - --nats=nats:4222
    ports:
      - 8021:8021
    depends_on:
      - dbs-api

  dbs-target-writer:
    container_name: target-writer
    image: slotix/dbs-target-writer
    entrypoint:
      - ./dbs-target-writer
      - --nats=nats:4222
      - --prometheus=http://prometheus:9090
    ports:
      - 8022:8022
    depends_on:
      - dbs-source-reader

  nats:
    container_name: nats
    image: nats
    entrypoint: /nats-server
    command: "--jetstream -m 8222 --store_dir /data/nats-server"
    ports:
      - 4222:4222
      - 8222:8222

  prometheus:
    image: slotix/dbs-prometheus:latest
    container_name: prom
    user: root
    ports:
      - 9090:9090

  mysql-source:
    container_name: mysql-source
    build: ./source 
    environment:
      - MYSQL_ROOT_PASSWORD=123456
    ports:
      - '3306:3306'


  postgres-target:
    container_name: postgres-target
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
    command: postgres
    ports:
      - '5432:5432' Docker-compose file for DBConvert Streams This docker-compose file is going to bring up multiple services and link them together so they can communicate with each other. With this setup, the replication process takes place between and databases, and it is controlled by , , and services. mysql-source postgres-target dbs-api dbs-source-reader dbs-target-writer nats DBConvert Streams services. service is the entry point of DBConvert Streams. It is where requests are sent with configuration settings for the source and target databases. It specifies the connection details to other components of the system such as , and . dbs-api source-reader target-writer service is responsible for monitoring and capturing changes in the source database, then sending batches of records to the Event Hub. dbs-source-reader service is used to receive changes from the Event Hub and apply them to the target database.. dbs-target-writer service is the core of the Event Hub, it provides communication between other DBS services. nats service is used for monitoring the metrics of DBS services. prometheus Database services. The structure of the tables that will be used in our example is depicted in the diagram below. The structure of MySQL source tables is adapted from the GitHub repository. jdaarevalo/docker_postgres_with_data Source and Target Databases. database image is based on , which includes all the necessary . This image also contains the which creates tables with the structures shown above. mysql-source slotix/dbs-mysql:8 settings to enable MySQL CDC replication initdb.sql script, database, on the other hand,  is based on the official lightweight image. database will receive all changes made to the database. postgres-target postgres:15-alpine postgres-target mysql-source Both of these databases, and , are typically located on separate physical servers in a production environment. However, in this example, we will run them on the same machine within distinct containers for demonstration purposes. mysql-source postgres-target Execution. To start all services described above, execute the following command: docker-compose up --build -d This command will use the file to build and start the necessary containers in detached mode (-d option). The flag will force the rebuilding of the images before starting the containers. docker-compose.yml --build To check if the MySQL database inside the running container has all tables created successfully by the script on start, you can run the command: docker exec -it mysql-source mysql -uroot -p123456 source -e 'SHOW TABLES;' By running the above command, you can see a list of tables created in the database inside the container, which will confirm if the script has created all the tables successfully on start. source +------------------+
| Tables_in_source |
+------------------+
| city             |
| country          |
| order_status     |
| product          |
| sale             |
| status_name      |
| store            |
| users            |
+------------------+ List of tables created on source Stream configuration This is the stream configuration file which is used to setup the database replication process. mysql2pg.json {
  "source": {
    "type": "mysql",
    "connection": "root:123456@tcp(mysql-source:3306)/source",
    "filter": {
      "tables": [
        { "name": "product", "operations": ["insert"]},
        { "name": "country", "operations": ["insert"]},
        { "name": "city", "operations": ["insert"]},
        { "name": "store", "operations": ["insert"]},
        { "name": "users", "operations": ["insert"]},
        { "name": "status_name", "operations": ["insert"]},
        { "name": "sale", "operations": ["insert"]},
        { "name": "order_status", "operations": ["insert"]}
      ]
    }
  },
  "target": {
    "type": "postgresql",
    "connection": "postgres://postgres:postgres@postgres-target:5432/postgres"
  },
  "limits": {
    "numberOfEvents": 0,
    "elapsedTime": 0
  }
} The field specifies the type of the source database as " " and the connection details to connect to the database, including username and password as well as the host and port. source mysql The field within the source field specifies that only certain tables and their corresponding operations (in this case " ") will be replicated. filter insert The field specifies the type of the target database as " " and the connection details to connect to the target database, including username and password as well as the host and port. target postgresql The field specifies that number of events and elapsed time are set to zero (0), which means that there are no limits in place for this replication process. limits It is also worth noting that according to this config, only insert operations will be captured on the tables specified in the filter field. Send configuration to DBConvert Streams API. docker run -t --rm \
    --network sales-db_default \
    curlimages/curl  \
    --request POST \
    --url http://dbs-api:8020/api/v1/streams\?file=./mysql2pg.json This command runs a command to start a new container from the image. It specifies that the container should join the network named using option. docker run curlimages/curl sales-db_default --network This container will then run the command to make a HTTP POST request to the url . The URL contains an endpoint that is the dbs-api service which is running on port 8020 and is expecting a json file as a query parameter. This command creates a new stream on the dbs-api service with the configuration specified in the file. curl http://dbs-api:8020/api/v1/streams?file=./mysql2pg.json mysql2pg.json It's important to note that this command assumes that the network and the service are already created and running. It also assumes that the file is in the current working directory from which the command is run. sales-db_default dbs-api mysql2pg.json This is a JSON response indicating that the stream creation was successful. {"status":"success",
	"data":{
        "id":"2KGlt8BCHLT0lXklrs5wqM6n7BQ",
        "source":{...},
        "target":{...},
        "limits":{}
     }
} It contains the following fields: : This field indicates the status of the request, in this case " " status success : This field contains the details of the stream that was created. data : This field contains a unique identifier for the stream, in this case "2KGlt8BCHLT0lXklrs5wqM6n7BQ" id : This field contains the details of the source database, including the type, connection details, and filter settings. source : This field contains the details of the target database, including the type and connection details. target : This field contains the limits for the replication process, such as number of events and elapsed time. limits Note that the details of the source and target field are not given here for brevity, it is just shown as ... Check if tables on the target are created successfully. 💡DBConvert Streams creates tables with the same structure as the source on the target if they are missing. At this point, all tables specified in the filter should exist on the Postgres target database. To connect to the Docker container and check if tables exist, you can run the following command: postgres-target docker exec -it postgres-target psql -U postgres -d postgres -c '\dt' By running the above command, you can see a list of tables created in the database, which will confirm if DBConvert Streams has created all the tables successfully. postgres-target List of relations
 Schema |     Name     | Type  |  Owner
--------+--------------+-------+----------
 public | city         | table | postgres
 public | country      | table | postgres
 public | order_status | table | postgres
 public | product      | table | postgres
 public | sale         | table | postgres
 public | status_name  | table | postgres
 public | store        | table | postgres
 public | users        | table | postgres
(8 rows) Populate the source with sample data. Now that the and databases have identical table sets with the same structure, it is time to find out if the streaming of data works properly. This can be done by inserting data into the database and observing if the same data is replicated to the database. mysql-source postgres-target mysql-source postgres-target -- Set params
SET @number_of_sales = '100';
SET @number_of_users = '100';
SET @number_of_products = '100';
SET @number_of_stores = '100';
SET @number_of_countries = '100';
SET @number_of_cities = '30';
SET @status_names = '5';
SET @start_date = '2023-01-01 00:00:00';
SET @end_date = '2023-02-01 00:00:00';

USE source;

TRUNCATE TABLE city ;
TRUNCATE TABLE product ;
TRUNCATE TABLE country ;
TRUNCATE TABLE status_name;
TRUNCATE TABLE users;
TRUNCATE TABLE order_status;
TRUNCATE TABLE sale;
TRUNCATE TABLE store;

-- Filling of products
INSERT INTO product
WITH RECURSIVE t(id) AS (
    SELECT 1
    UNION ALL
    SELECT id + 1
    FROM t
    WHERE id + 1 <= @number_of_products
)
SELECT id, CONCAT_WS(' ','Product', id)
FROM t;


-- Filling of countries
INSERT INTO country
WITH RECURSIVE t(id) AS (
    SELECT 1
    UNION ALL
    SELECT id + 1
    FROM t
    WHERE id + 1 <= @number_of_countries
)
SELECT id, CONCAT('Country ', id)
FROM t;


-- Filling of cities
INSERT INTO city
WITH RECURSIVE t(id) AS (
    SELECT 1
    UNION ALL
    SELECT id + 1
    FROM t
    WHERE id + 1 <= @number_of_cities
)
SELECT id
    , CONCAT('City ', id)
    , FLOOR(RAND() * (@number_of_countries + 1))
FROM t;


-- Filling of stores
INSERT INTO store
WITH RECURSIVE t(id) AS (
    SELECT 1
    UNION ALL
    SELECT id + 1
    FROM t
    WHERE id + 1 <= @number_of_stores
)
SELECT id
    , CONCAT('Store ', id)
    , FLOOR(RAND() * (@number_of_cities + 1))
FROM t;

-- Filling of users
INSERT INTO users
WITH RECURSIVE t(id) AS (
    SELECT 1
    UNION ALL
    SELECT id + 1
    FROM t
    WHERE id + 1 <= @number_of_users
)
SELECT id
    , CONCAT('User ', id)
FROM t;

-- Filling of status_names
INSERT INTO status_name
WITH RECURSIVE t(status_name_id) AS (
    SELECT 1
    UNION ALL
    SELECT status_name_id + 1
    FROM t
    WHERE status_name_id + 1 <= @status_names
)
SELECT status_name_id
    , CONCAT('Status Name ', status_name_id)
FROM t;

-- Filling of sales  
INSERT INTO sale
WITH RECURSIVE t(sale_id) AS (
    SELECT 1
    UNION ALL
    SELECT sale_id + 1
    FROM t
    WHERE sale_id + 1 <= @number_of_sales
)
SELECT UUID() AS sale_id
    , ROUND(RAND() * 10, 3) AS amount
    , DATE_ADD(@start_date, INTERVAL RAND() * 5 DAY) AS date_sale
    , FLOOR(RAND() * (@number_of_products + 1)) AS product_id
    , FLOOR(RAND() * (@number_of_users + 1)) AS user_id
    , FLOOR(RAND() * (@number_of_stores + 1)) AS store_id
FROM  t;

-- Filling of order_status
INSERT INTO order_status
WITH RECURSIVE t(order_status_id) AS (
    SELECT 1
    UNION ALL
    SELECT order_status_id + 1
    FROM t
    WHERE order_status_id + 1 <= @number_of_sales
)
SELECT UUID() AS order_status_id
    , DATE_ADD(@start_date, INTERVAL RAND() * 5 DAY) AS update_at
    , FLOOR(RAND() * (@number_of_sales + 1)) AS sale_id
    , FLOOR(RAND() * (@status_names + 1)) AS status_name_id
FROM t; fill_tables.sql script This script above starts by truncating all the tables in the database to clear any previous data, and then it inserts sample data into the tables. Each table is filled with sample data using a different set of parameters set at the top of the script. source To execute this SQL script, you can run the following command: docker exec -i \
    mysql-source \
    mysql -uroot -p123456 -D source < $PWD/fill_tables.sql Comparing number of records in source and target databases. Let's compare the number of rows in the tables of the source and target databases. docker exec -it mysql-source mysql -uroot -p123456 -D source -e "SELECT (SELECT COUNT(*) FROM product) as 'product_count',(SELECT COUNT(*) FROM country) as 'country_count',(SELECT COUNT(*) FROM city) as 'city_count',(SELECT COUNT(*) FROM store) as 'store_count',(SELECT COUNT(*) FROM users) as 'users_count',(SELECT COUNT(*) FROM status_name) as 'status_name_count',(SELECT COUNT(*) FROM sale) as 'sale_count',(SELECT COUNT(*) FROM order_status) as 'order_status_count';" How many records are in each table of mysql-source db? docker exec -it postgres-target psql -U postgres -d postgres -c "SELECT (SELECT COUNT(*) FROM product) as product_count,(SELECT COUNT(*) FROM country) as country_count,(SELECT COUNT(*) FROM city) as city_count,(SELECT COUNT(*) FROM store) as store_count,(SELECT COUNT(*) FROM users) as users_count,(SELECT COUNT(*) FROM status_name) as status_name_count,(SELECT COUNT(*) FROM sale) as sale_count,(SELECT COUNT(*) FROM order_status) as order_status_count;" how many records are in each table of postgres-target db? As you can see from the resulting output, all tables in both the source and target databases have an identical number of records in each table. Check statistics. The following command sends a GET request to the DBConvert API endpoint which retrieves the statistics of the current data stream. The command is used to format the JSON output for better readability. /api/v1/streams/stat jq docker run -t --rm \
    --network sales-db_default \
    curlimages/curl  \
    --request GET \
    --url http://dbs-api:8020/api/v1/streams/stat | jq {
  "streamID": "2KHTjpsZAUCb8y1BZny3YKoX5qO",
  "source": {
    "counter": 635,
    "elapsed": "0s",
    "started": "2023-01-13T17:24:50.089257721Z",
    "status": "RUNNING"
  },
  "target": {
    "counter": 635,
    "elapsed": "0s",
    "started": "2023-01-13T17:24:50.089649312Z",
    "status": "RUNNING"
  }
} The above output shows the statistics of the current data stream. The "source" field shows the statistics of the source database, and the "target" field shows the statistics of the target database. The "counter" field shows the number of events that have been processed by the stream, the "elapsed" field shows the time elapsed since the stream started, the "started" field shows the date and time when the stream was started, and "status" field shows the status of the current stream. Prometheus metrics. Prometheus is a monitoring system that scrapes metrics data from various sources and stores them in a time-series database. DBConvert Streams collects its internal metrics in Prometheus format, allowing you to explore and visualize live data in dashboards. To view the collected metrics, visit in a web browser to access the Prometheus UI. http://127.0.0.1:9090 Conclusion Change Data Capture (CDC) systems like can be used to stream data from a MySQL database to a PostgreSQL database in real time, allowing you to keep the two systems in sync and take advantage of the unique features and capabilities of each database. DBConvert Streams This guide provided information on streaming data in one direction, from MySQL Binlog to PostgreSQL. contains more examples of configuring data streams, including from PostgreSQL Wals to MySQL and other configurations. These examples can serve as a starting point for setting up your own data stream tailored to your specific needs and use case. The DBConvert Streams Github repository Getting your feedback about DBConvert Streams is essential for the development team to improve the software and make it more useful for the community. By sharing your ideas, reporting bugs, and requesting new features, you can actively participate in the development of the software and help to make it more robust and useful for everyone. Your participation is valued and appreciated by the development team and the community. Also Published Here

How to Stream Data from MySQL to Postgres

In today’s market, hardware alone isn’t enough to make a smartphone stand out. Over the years, there have been countless smartphone models that didn’t take off or didn’t do enough to stand out among the competition — such as Amazon’s Fire Phone and the HTC First. Of course, the battle for unique identity and the sales that go with it don’t just apply to the smartphone market — that’s why makers of cars, thermostats and just about every other device are adding “smarts” to their products to differentiate them from the pack. This phenomenon highlights what’s becoming a clearer reality in today’s world. In order to survive, the smartphone will have to become more than the rectangle device that people hold in their hands — it will have to shed its physical form and become the cloud. The Secret to Better Hardware: Software and Services During his F8 conference where he discussed new technology for smartphones, Facebook CEO Mark Zuckerberg said, “We aren’t using primitive tools because we prefer primitive tools. We are using primitive tools because we are still early in the journey in developing better tools.” Smartphones have amazing potential in ways we haven’t even dreamed about yet. For now, we’re still dealing with technological limitations — especially hardware limitations. As apps continue to get more powerful, they’ll also cost a smartphone more resources, like battery and data; according to researchers with Duke, UCLA and other organizations, the . keynote speech battery remains the primary bottleneck for mobile devices To overcome these constraints, we need to take a holistic approach to technology. Add software and services to the equation — because software coupled with hardware will beat hardware alone every time. Software works around the battery bottleneck because software doesn’t have the physical limitations of hardware. Why have hardware do the task alone, when it can team up with software and services to do the job easier and better? Think of how cars have developed over the years — from diesel to unleaded gasoline to electricity, we’ve refined the fuel that goes in the car, not the car itself. A car still has four wheels, an engine and a steering wheel. A smartphone’s hardware is the car itself — you can’t reinvent the wheel, and you can’t realistically expand what a smartphone is either. Software is the destination — where you want to take the car. What we need to change is the fuel — in this case, the processing. Processing on the phone itself is diesel — cloud processing is nitro-boosted gasoline. Choosing a Server Room over a Single Processor No matter how great a phone is, it’s still a single processor, and limited by what one processor can do. Having an entire server room at your beck and call is far better. The challenge with mobile computing has always been the tradeoff between portability and power. Because it’s hard to fit a server room in your pocket, users have foregone additional processing power to take their computing on the go. But with cloud processing, that tradeoff can finally be part of the smartphone’s history. Cloud processing acts as a direct link between your phones and the server room. Your phone is no longer just your smartphone: it’s also the server room and the cloud connection binding them together. That’s hardware plus software plus services in action, effectively transforming a smartphone into the cloud. One of the primary benefits of this will be speed. A room of servers is going to be able to handle information much quicker than a single processor. Browsing speed benchmarks have been an often-used comparison into the processing capabilities of these programs, and they show the difference cloud processing makes. In the widely cited JetStream benchmark, cloud processing scored a 210, while processing through the smartphone itself scored at most a 63. This also shows why more emphasis should be given to software and services rather than hardware — the physical phone itself made little difference. Researchers with the have concluded that cloud computing leads to “explicit resource saving” by speeding up the execution of apps and services with no loss of quality — particularly saving local CPU use and battery and data consumption. For example, a combination of cloud computing technology and JavaScript can be used to preprocess and compress web pages, achieving both faster loading times and browsing that can save up to 90 percent of a smartphone’s data bandwidth. The only processing happening on a user’s smartphone is limited — most of the action is happening on separate cloud servers, so it wouldn’t be charged to a user’s data plan. University of Malaysia In addition, cloud processing holds great benefits for a smartphone’s security. By connecting to the internet through cloud processing, a user is effectively kept at arm’s distance from hackers and internet-based threats. Even if attackers manage to breach a connection (which for maximum security, should be encrypted), they will be accessing the cloud servers, not the personal device of a user. If these cloud servers are used for only processing and not to store data, attackers will be left disappointed by only being able to reach servers with no data to mine. Finding the Solution in the Cloud Taking a smartphone’s processing to the cloud isn’t a drastic step and is something we can implement right now. We’ve already started the process: Facebook, Google Drive, Salesforce and Microsoft Office 365 are just some examples of cloud services that we use every day. You effectively take the philosophy and function that fuels Software-as-a-service (SaaS) platforms, and you apply it to every aspect of what a phone does — Browsing-as-a-service, Gaming-as-a-service, and so on. The time to begin these transitions is now. It may seem like the heir apparent to smartphones is a long way off, but if smartphones don’t begin their transcendence to the cloud soon, they’ll be left behind when the next hardware device comes to replace it, whatever that may be. History is littered with devices that came and went, replaced by the next gadget. This is what awaits smartphones if they remain a pure hardware device. This is why the computer had such an impact — it became the network and grew to have an impact beyond its CPU, monitor, mouse and keyboard. It represented connection and engagement like the world had never seen before. The smartphone must now mirror it by becoming the cloud to reach faster functionality. By connecting everyday products, we will be in the age of devices made by people who understand that hardware, software and services in one package make up the winning formula. Smartphones must embrace this concept too, and the cloud will be the way to do that. Shioupyn ShenPresident & CEO This post was originally published on IoT Central.

Why the Future of Mobile Technology Lies in the Cloud

I just watched ’s it addresses the issue of concurrency collisions that can occur in large complicated systems due to the increase in the number and complexity of ajax calls being made. He has offered an alternative solution with Logux. Andrey Sitnik talk on Logux The coordination of many client actions to many servers into a single data store is a daunting task. Maintaining the synchronicity of changes being made to the data in the correct order offers a host of problems. Andrey points out that the historical problem exists between the state-machine running on the client and the MVC running on the server. The no man’s land in between has been largely navigated using AJAX calls. As systems grow and features get added these calls to the server become tedious and prone to error not to mention chatty. This is where Logux comes in. Log files exist both on the server and on the client. We log all kinds of actions on the client and every scrap of execution on the server. The premise of is to make the log a first class citizen of the application. In my mind, unifying the logging of all aspects of the application nails down the tenants of as outlined by Martin Fowler. @Logux_io event sourcing To ultimately use it as a source of truth for the activities passed between the client and the server. This will all occur using the immutable update pattern that Redux implements. It seems to me to naturally extend Redux to have a artifact useful for the coordination of the system itself. The log will be well defined and each action within the stack will have an atomic identifier outlined by the paper published in 2009. CRDTs: Consistency Without Concurrency Control With the concept of self ordering atomic actions baked in for both the client and the server then it would become trivial to commit those actions in the order that they occur beit real time or when coming back online. There are a number of challenges that have been at least roughly addressed concerning the whole realm of date-time gotchas that almost every large scale application has to face anyway. Andrey seems to have narrowed that down and ironed out the details concerning the agreement and coordination of those key points as well as handling the dead on the line scenario by maintaining a heart beat between the server and application. All in all, it was a great talk by one of the . EvilMartians Cheers! Here are some obligatory links: & & The Logux GitHub Shout Out to the Amazing Martin Fowler ThoughtWorks Andrey Sitnik EvilMartians

Logux: Alternate approach to using AJAX

, we discussed what event streaming windowing is, and we examined in detail the structure of a windowed aggregate in Kafka Streams and Flink SQL. In the first post of this series In this post, we'll dive into two specific windowing implementations: hopping and tumbling windows. Hopping Windows A hopping window has a fixed time length, and it moves forward or "hops" at a time interval smaller than the window's length. For example, a hopping window can be one minute long and advance every ten seconds. The following illustration demonstrates the concept of a hopping window: So, from the illustration above, hopping windows can produce overlapping results. A hop forward can include results contained in the previous window. Let's look at another illustration demonstrating this concept: Walking through the picture: Window one starts at 12:00:00 PM and will collect data until 12:01:00 PM (end time is exclusive). At 12:00:30 PM, due to the thirty-second advance, window two starts gathering data. Window one and window two will share data for thirty seconds from the start of window two until the end of window one. The process continues with each window advance. Let's show how you would implement a hopping window in Kafka Streams and Flink SQL. Kafka Streams Hopping Window For a hopping windowed aggregation in Kafka Streams, you'll use one of the factory methods in the class: TimeWindows KStream<String,Double> iotHeatSensorStream =
  builder.stream("heat-sensor-input",
    Consumed.with(stringSerde, doubleSerde));

iotHeatSensorStream.groupByKey()
      .windowedBy(
                  TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)) <1>
                             .advanceBy(Duration.ofSeconds(30)) <2>
                  )
        .aggregate(() -> new IotSensorAggregation(tempThreshold),
         aggregator,
         Materialized.with(stringSerde, aggregationSerde))
         .toStream().to("sensor-agg-output",
           Produced.with(windowedSerde, aggregationSerde)) By using sets the window size at one minute, the means Kafka Streams will drop any out-of-order records that would have been included in the window had they arrived in order. We'll get into grace periods more in the blog post on windowing time semantics. TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)) withNoGrace The call makes this a hopping window. It creates a window that is one minute in size and advances every ten seconds. .advanceBy(Duration.ofSeconds(30) Next, let's move on to hopping windows with Flink SQL. Flink SQL Hopping Window Note that Flink hopping windows can also be referred to as sliding windows. Kafka Stream offers a sliding window variant that behaves differently from its hopping window offering. So, for clarity, we'll only refer to Flink windows with an advance smaller than the window size as hopping windows. SELECT window_start,
       window_end,
       device_id,
       AVG(reading) AS avg_reading
FROM TABLE(HOP <1>
               (TABLE device_readings,   <2>
                     DESCRIPTOR(ts),    <3>
                     INTERVAL '30' SECONDS,  <4>
                     INTERVAL '1' MINUTES  <5>
               ))
GROUP BY window_start,
         window_end,
         device_id Specifying hopping windows by passing the function to the function. HOP TABLE The table you'll use as the source for the hopping window aggregation. The is the column with the time attribute used for the window. DESCRIPTOR This first is the amount of "hop" or advance of the window. INTERVAL The second is the size of the window. INTERVAL Now, let's move on to tumbling windows. Tumbling Windows A tumbling window has a fixed length in size and has an advance that is the same amount of time as the size. Tumbling windows are considered a specialized case of hopping windows due to the advance equalling the window size. Since a tumbling window starts a new one when the previous one ends, they don't share any data. You won't find records from one window in another one; the following illustration helps clarify this process: Stepping through this illustration Window one starts at 12:00:00 PM and will collect data until it ends. The end time of the window is exclusive. At 12:01:00 PM window two starts collecting data. Since each window starts collecting data after the previous window ends, there are no shared results. Kafka Streams Tumbling Window KStream<String,Double> iotHeatSensorStream =
  builder.stream("heat-sensor-input",
    Consumed.with(stringSerde, doubleSerde));

iotHeatSensorStream.groupByKey()
      .windowedBy(
                  TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)) <1>
                  )
        .aggregate(() -> new IotSensorAggregation(tempThreshold),
         aggregator,
         Materialized.with(stringSerde, aggregationSerde))
         .toStream().to("sensor-agg-output",
           Produced.with(windowedSerde, aggregationSerde)) Using without the clause automatically makes this a tumbling window. Since you didn't specify an advance, Kafka Streams will add one equal to the size. TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)) .advanceBy You could add the clause with the same amount of time if you choose to skip the shortened version. advanceBy Flink SQL Tumbling Window Tumbling windows in Flink SQL are defined similarly to the hopping variety with a couple of differences. SELECT window_start,
       window_end,
       device_id,
       AVG(reading) AS avg_reading
FROM TABLE(TUMBLE <1>
               (TABLE device_readings,   <2>
                     DESCRIPTOR(timestamp),    <3>
                     INTERVAL '1' MINUTES  <4>
               ))
GROUP BY window_start,
         window_end,
         device_id Specifying tumbling windows by passing the function to the function. TUMBLE TABLE The table you'll use as the source for the tumbling window aggregation. The is the column with the time attribute used for the window. DESCRIPTOR By passing a single parameter, Flink SQL will utilize this as the advance and the size. INTERVAL You define tumbling windows in Flink SQL similarly to Kafka Streams in that you only provide a one-time parameter. Use Cases Let's look at an illustration for the use case of hopping windows: So, from looking at this image, we could generalize hopping windows as "every <time-period of window advance> give me <aggregate> over the last <window size> period."  From our examples here, it would be "every 30 seconds, give me the average temp reading over the last minute." So, any problem domain where you want to closely monitor changes over time or compare changes relative to the previous reading could be a good fit for the hopping window. It’s worth noting that Kafka Streams will emit periodic results before a window closes, while Flink SQL will only emit results when it’s closed. I’ll go into more detail about this in the fourth installment of this blog series. For the tumbling window, we have another illustration: This illustration for tumbling windows can be summarized as "give me <aggregate> every <window size period>" or restated to fit the examples in this post: "give me the average temp reading over the last minute." Since tumbling windows don't share any records, any situation requiring a unique count of events per window period would be a reason to use a tumbling window. Resources Apache Flink® on Confluent Cloud Flink SQL Windows Kafka Streams windowing documentation Kafka Streams in Action 2nd Edition

How to Master Stream Processing - Hopping and Tumbling Windows

Stream processing is the best way to work with event data. While batch processing still has its use cases, and probably always will, only stream processing offers the ability to respond in real time to events. But if we zoom in, what does it look like to respond to events? By now, I’m sure you’re familiar with the oft-quoted fraud scenario - a person with nefarious intent gets a hold of an unaware consumer’s credit card number. Still, due to the bank’s responsiveness processing system, the fraudulent charge gets declined. Other uses of stream processing require an immediate response but are not tied to one single event. Consider monitoring the heat of a manufacturing process; if the average temperature reaches a certain threshold in a given period, then the monitoring process should generate an alert. But this isn’t about one temperature spike. It’s about a consistent upward trend. In other words, what are the temperature readings doing during a fixed period? I’m talking about windowing in event streams if you have not guessed by now. While aggregations (an aggregation is a grouping of events by a common attribute) are a vital tool to leverage an event stream, an aggregation over all time doesn’t shed any light on specific periods of activity. Consider the following illustration: Over time the average temperature reading has increased some over time, but it doesn’t tell the whole story. Now let’s take a look at capturing the average temp readings over specific intervals: Now, by getting readings at specific intervals (windows), you can spot the issue with a large jump in the average value. This is not to say that an aggregation over all time isn't helpful, but
that, in many cases, you'll want to aggregate over specific intervals.
In other cases, you'll want an aggregation not defined by fixed time
boundaries but by behavior, e.g., session windows whose boundaries are
based on periods of We'll get into session windows in a
post later in the blog series. inactivity. This blog post marks the first in a series about windowing in the two dominant stream processing technologies today: and Kafka Streams Flink, specifically Flink SQL It's important to note that the point of this blog series is not a direct comparison between the two APIs. Instead, it is a resource for windowed operations in Kafka Streams and Flink SQL. While comparing the two in a competitive analysis is natural, it's not the main focus here. The blog series will discuss: The different types of windowing, semantics, and potential use cases. Time semantics Interpretation of the results Testing windowed applications I will assume basic familiarity with Kafka Streams and Flink SQL, so the examples will start by covering windowing. But before we get into windowing, let’s discuss how Kafka Streams and Flink SQL structure windowing applications. We’ll only cover this level of detail in this initial post, and subsequent ones will assume knowledge of how to assemble the program and focus on the windowing aspect. Kafka Streams windowing You’ll need to specify an to do any windowing in Kafka Streams. Aggregations are a function that combines smaller components into a large composition, clustered around some attribute, which in Kafka Streams will be the key in the key-value pairs. You can also perform a reduce, a specialized form of aggregation since a reduce operation will return the same type as its input components. Generally, an aggregation can return a completely different value from the inputs. But since windowing operates the same for either a reduction or aggregation, we will use an aggregation for our examples throughout the blog series. aggregation KStream<String,Double> iotHeatSensorStream =
  builder.stream("heat-sensor-input",
    Consumed.with(stringSerde, doubleSerde));
iotHeatSensorStream.groupByKey() <1>
      .windowedBy(<window specificatation>) <2>
        .aggregate(() -> new IotSensorAggregation(tempThreshold), <3>
         aggregator,
         Materialized.with(stringSerde, aggregationSerde))
         .toStream().to("sensor-agg-output",
           Produced.with(windowedSerde, sensorAggregationSerde))<4> Let's walk through the essential points of setting up the Kafka Streams
window aggregation: The first step is to group all records by key; this is required before performing any aggregation. Here, you're using , which assumes the underlying key-value pairs have the correct keys needed for clustering together. If not, you could use the function, where you pass a instance that maps the current key-value pair into a new one, which allows you to create a new key suitable for the aggregation grouping. Note that changing the key for a group-by will lead to a re-partitioning of the records. KStream.groupByKey KStream.groupBy KeyValueMapper You are specifying the windowing - we'll cover the specific types in
later posts. Point three is where you're specifying how to aggregate records. The
first parameter is an Initializer represented as a lambda function,
which provides the initial value. The second parameter is the
Aggregator instance, which performs the aggregation action you
specify. Here, it's a simple average and tracking the highest and
lowest values seen. The third parameter is a Materialized instance
specifying how to store the aggregation. Since the value type
differs from the incoming value, you must provide the appropriate
Serde instance for Kafka Streams to use when (de)serializing
records. The final point is where you provide the Serde instances for producing the results back to Kafka. The key Serde is a different type, as Kafka Streams wraps the incoming record key in a Windowed instance. What's not apparent from this aggregation example is where the
timestamps for the window are. But there's a big hint in the explanation
of the aggregation example. At point four of the aggregation
description, Kafka Streams wraps the original key in a object. Windowed As shown in this illustration, the Windowed object contains the original key and the instance for the aggregation values. The Window object has the start and
end time for the aggregation window. It doesn't contain the window size,
but you can easily calculate the size by subtracting the start time from
the end. We'll cover reporting and analyzing the aggregation window
times in a follow-on blog post. Window Wrapping the original key in a Windowed object changes the type, meaning you'll have to update Kafka Streams on serializing the results. Fortunately, Kafka Streams provides the utility class, making it easy to get the correct Serde for producing results back to Kafka: WindowedSerdes Serde<Windowed<String>> windowedSerde =
        WindowedSerdes.timeWindowedSerdeFrom(String.class, <1>
                                              60_000L <2>
                                            );

KStream<String,Double> iotHeatSensorStream =
  builder.stream("heat-sensor-input",
    Consumed.with(stringSerde, doubleSerde));
  iotHeatSensorStream.groupByKey() <1>
         .windowedBy(<window specificatation>)
         .aggregate(() -> new IotSensorAggregation(tempThreshold),
              aggregator,
               Materialized.with(stringSerde, aggregationSerde))
         .toStream().to("sensor-agg-output",
           Produced.with(windowedSerde, sensorAggregationSerde)) <3> The class type for the original key The size of the window in milliseconds Providing the Serde for the Windowed key So, by using the WindowedSerdes class, you provide the proper
deserialization strategy for Kafka Streams to produce windowed results
back to Kafka. Producing windowed results to a topic implies downstream
consumers will know how to handle the windowed results as well. We'll
cover that situation in a later blog on reporting in a subsequent post
in this series. Now, let's move on to Flink SQL aggregation windows. Flink SQL windowing Flink offers windowing for event stream data as windowing table-valued
functions (TVF). The Flink TVFs implement the (PTF). In a nutshell, PTFs allow for user-defined functions on a table
that returns a table. SQL 2016 standard
Polymorphic Table
Functions The exciting thing about PTF is that the schema of the table returned by the function is dynamic; it's determined at runtime by the function output. So, the PTFs enable windowing and aggregation functions on existing tables, which is precisely what we get with the Flink SQL windowing. The windowing TVFs in Flink replace the now deprecated .
Window TVFs provide more powerful window-based calculations like and . Group Window Functions Window
TopN Window
Deduplication Now, let's move on to how you execute a windowed aggregation in Flink
SQL. As with the Kafka Streams example, we'll review the structure of a
windowed aggregation, with specific window implementations covered in
later posts. SELECT window_start,
       window_end,
       device_id,
       AVG(reading) AS avg_reading   <1>

FROM TABLE(<2>
           <Window Function> ( <3>
                              TABLE device_readings,   <4>
                              DESCRIPTOR(ts),    <5>
                              INTERVAL '5' MINUTES,  <6>
                              [INTERVAL '10' MINUTES]
                            )
           )
GROUP BY window_start, <7>
         window_end,
         device_id Here's the breakdown of the query: Selecting the columns and the aggregation using the Flink SQL AVG
function and providing a descriptive name; these columns form the
schema of the returned table. The TABLE function Here, you give a specific window function, either HOP, TUMBLING, or
CUMULATE. Support for a SESSION type is coming soon. We'll cover the
specific types in later posts. Next are the parameters for the window function, starting with the
table to use for the input The DESCRIPTOR is the time attribute column the function uses for
the window. Depending on the window function, the following 1 or 2 parameters
determine the window advance and size or just the size. As with standard SQL aggregate functions, we need the same columns
in the GROUP BY clause in the SELECT clause. Flink SQL inserts three additional columns into windowed operations: window_start, window_end, and window_time. Flink SQL determines window_time by subtracting 1ms from the window_end value. This concludes our introduction to the structure of windowing
applications in Kafka Streams and Flink SQL. In the next edition, we'll
cover hopping and tumbling windows. Resources Apache Flink on Confluent
Cloud Flink SQL
Windows Kafka Streams windowing
documentation Kafka Streams in Action 2nd
Edition

Mastering Stream Processing: A Guide to Windowing in Kafka Streams and Flink SQL

Converting Excel or CSV files into Java objects (POJOs) and vice versa can be a complex process, but with the right tools and techniques, it becomes much more manageable. In this guide, we’ll explore a powerful library that leverages Java reflection. Java First, we add the dependency to Maven. <dependency>
  <groupId>com.adnanebk</groupId>
  <artifactId>excel-csv-converter</artifactId>
  <version>0.0.1-SNAPSHOT</version>
</dependency> Understanding the POJO Class Before we dive into the library, let’s take a close look at a sample Java class that serves as our data model: @Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@SheetDefinition(datePattern = "dd/MM/yyyy")
public class Product {

    @CellDefinition(0)
    private String name;

    @CellDefinition(1)
    private long price;

    @CellDefinition(2)
    @CellBoolean(trueValue = "yes",falseValue = "no")
    private boolean active;

    @CellDefinition(value = 3, title = "Promo price")
    private double promoPrice;

    // Additional fields...

    @CellEnum(enumsMapperMethod = "categoryMap")
    @CellDefinition(10)
    private Category category;

    @CellDefinition(11)
    private LocalDateTime localDateTime;

    private Map<Category,String> categoryMap(){
        return Map.of(Category.A,"Formatted A",
                      Category.B,"Formatted B");
    }
} This class is annotated with various annotations that play a crucial role in the conversion process. Each field is annotated with , indicating its position in the . Product @CellDefinition Excel or CSV file we can also define the title of the of the cell, by default, it will convert the camel case name of the field to a name with spaces (ex: firstName=>First name) The annotation provides additional information like date formatting patterns that will be used during the conversion of date field types. @SheetDefinition The Enum Annotation: @CellEnum(enumsMapperMethod = “categoryMap”) In the class, we make use of the annotation in the enum Category field. The enumsMapperMethod argument allows us to define a method name; this method should return a map that defines the mapping (conversions) between the enum constants and the formatted values in the Excel/CSV cells (by default, the enum constants will be used), note that the method name must much the enumsMapperMethod argument value. Product @CellEnum The Boolean Annotation: @CellBoolean(trueValue = “yes”,falseValue = “no”) we make use of the @CellBoolean annotation in the boolean active field, it has two arguments that represent the formatting values we want to use in the Excel/CSV fields. Now, let’s introduce an updated version of our POJO class, : ProductV2 @Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@SheetDefinition(includeAllFields = true,titles={"Name","Category","Date"})
public class ProductV2 {

    private String name;

    // Additional fields...

    private Category category;

    @IgnoreCell
    private LocalDateTime localDateTime;
} Converting Excel files to POJOs becomes even simpler with the annotation provided in the class to map the fields to corresponding cells in the Excel file. @SheetDefinition When the includeAllFields argument is set to true, the fields are automatically included and mapped in the cells based on their declared order and ignore fields that are annotated with @IgnoreCell annotation. We can define the titles in the titles argument with the condition that they must be in the same order as the fields. Converting Excel/CSV to POJOs and vice versa @RestController
@RequestMapping("excel/products")
public class ExcelFieldsController {
    private final ExcelHelper<Product> excelHelper = ExcelHelper.create(Product.class);

    @GetMapping
    public List<Product> excelToProducts(@RequestBody MultipartFile file){
        return excelHelper.toStream(file.getInputStream()).toList();
    }

       @GetMapping("/download")
    public ResponseEntity<InputStreamResource>
    downloadExcelFromProducts() {
        String filename = "products-" + LocalDate.now() + ".xlsx";
        InputStreamResource file = new InputStreamResource(excelHelper.toExcel(getProducts()));
        return ResponseEntity.ok()
                .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=" + filename)
                .contentType(MediaType.parseMediaType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"))
                .body(file);
    }
} the same is applicable for converting CSV files, except we need to define the delimiter that will be used private final CsvHelper<ProductV2> csvHelper = CsvHelper.create(ProductV2.class,";"); The Class: Dynamic Class Examination ReflectionUtil The ReflectionUtil class serves as the backbone of this Java library, facilitating dynamic class examination and manipulation through the power of Java reflection. One noteworthy feature of the ReflectionUtil class is the optimization applied to enhance performance. During initialization, all getters, setters, and fields are eagerly loaded and encapsulated in the SheetField record. This deliberate action minimizes the need for reflection lookups in subsequent operations and boosts overall efficiency. Record Overview SheetField The Field record is a fundamental component of the library, designed to encapsulate information about the field of a class. public record SheetField<T>(String typeName, String title, Function<T,Object> getter, BiConsumer<T,Object> setter, int cellIndex) Key methods : : Retrieves the value of the field from an object using its getter method. If the field is an enum, it provides formatted values based on defined enum mappings. public Object getValue(T obj) : Sets the value of the field in an object using its setter method. It handles enum values and ensures proper conversion. public void setValue(T obj, Object value) Conclusion By leveraging this custom library, developers can significantly simplify the process of converting Excel and CSV files to . The integration of Java reflection, along with thoughtful design considerations, empowers dynamic mapping, making it a valuable tool for data processing tasks. POJOs in Java GitHub repo https://github.com/adnanebk/excel-pojo-converter

Simplify Data Conversion: Excel and CSV to Java Objects with Reflection

On-Demand Live Stream Playback with Amazon IVS (3 Part Series) 1 Auto Recording Amazon IVS Live Streams to S3 2 How to Archive Web Chat Messages with Amazon IVS Chat Logging 3 Amazon IVS Live Stream Playback with Chat Replay In our , we looked at how to auto-record Amazon Interactive Video Service (Amazon IVS) live streams to an Amazon Simple Storage Service (Amazon S3) bucket. That feature is a powerful tool for user generated content (UGC) platforms as it is the first step in providing on-demand viewing of previous live streams in an application. But a replay of a live stream is incomplete without the full context of the interactive chats that occur alongside of the stream. In this post, we'll take the next step to providing a full on-demand experience by learning how to log Amazon IVS chat messages. last post Logging Configurations Similar to how auto-recording to Amazon S3 required a recording configuration, chat logging also requires a configuration that defines the necessary details for a chat room. Keep in mind that you might also need additional depending on your choice of logging destinations. IAM permissions Using the Amazon IVS Console To Create a Logging Configuration To create a logging configuration with the Amazon IVS Console, click on in the left sidebar. Logging Configurations On the list page, click . Logging configurations Create logging configuration Amazon IVS chat logging offers several destinations for storage of the logged chat messages. We can choose from Amazon CloudWatch, an Amazon Kinesis Data Firehose, or Amazon S3 as a storage destination for your chat logs. I personally find CloudWatch to be the most convenient destination since I can easily retrieve the chat messages via the CloudWatch SDK, so let's use that for this demo. Refer to the if you'd like to utilize an Amazon Kinesis Data Firehose or Amazon S3 as a destination in your application. documentation : There can be a slight delay between when a chat message is posted and when it appears at your logging destination. For Amazon S3, the delay can be up to 5 minutes, and for Amazon CloudWatch and Amazon Kinesis Data Firehose it can be up to 10 seconds. Note To create the configuration for this demo, enter a (#1), select CloudWatch as the (#2), select (#3), and enter a (#4). Logging configuration name Destination Create a new CloudWatch log group Log group name Enter any optional and then click . Tags Create logging configuration Using the AWS SDK To Create a Logging Configuration Just like with recording configurations, we can take advantage of the AWS SDK to create our logging configurations. Since we're logging to CloudWatch in this post, we'll need to use the CloudWatch Logs SDK to . create a log group import { CloudWatchLogsClient, CreateLogGroupCommand } from "@aws-sdk/client-cloudwatch-logs";

const cloudWatchLogsClient = new CloudWatchLogsClient();
const createLogGroupInput = {
  logGroupName: 'ivs-demo-chat-logging-group',
};
const createLogGroupRequest = new CreateLogGroupCommand(createLogGroupInput);
const createLogGroupResponse = await cloudWatchLogsClient.send(createLogGroupRequest);
console.log(createLogGroupResponse); Now we can use the Amazon IVS chat client to create a logging configuration ( ). docs import { IvschatClient, CreateLoggingConfigurationCommand } from "@aws-sdk/client-ivschat";

const ivsChatClient = new IvschatClient();
const createLoggingConfigInput = {
  name: 'ivs-demo-chat-logging-config-sdk',
  destinationConfiguration: {
    cloudWatchLogs: {
      logGroupName: 'ivs-demo-chat-logging-group'
    }
  }
};
const createLoggingConfigRequest = new CreateLoggingConfigurationCommand(createLoggingConfigInput);
const createLoggingConfigResponse = await ivsChatClient.send(createLoggingConfigRequest);
console.log(createLoggingConfigResponse); This will produce output similar to the following: {
  "$metadata": {
    "httpStatusCode": 200,
    "requestId": "[redacted]",
    "cfId": "[redacted]",
    "attempts": 1,
    "totalRetryDelay": 0
  },
  "arn": "arn:aws:ivschat:us-east-1:[redacted]:logging-configuration/[redacted]",
  "createTime": "2023-01-09T14:48:35.358Z",
  "destinationConfiguration": {
    "cloudWatchLogs": {
      "logGroupName": "ivs-demo-chat-logging-group"
    }
  },
  "id": "[redacted]",
  "name": "ivs-demo-chat-logging-config-sdk",
  "state": "ACTIVE",
  "tags": {},
  "updateTime": "2023-01-09T14:48:35.485Z"
} Creating a Logging Configuration with the AWS CLI We can also optionally create a logging configuration via the AWS CLI. Again, we'll have to create the CloudWatch log group, then pass the name of the new log group into the chat logging configuration. $ aws logs \
  create-log-group \
  --log-group-name ivs-demo-logging-config-log-group

$ aws ivschat \
  create-logging-configuration \
  --name ivs-demo-logging-config \
  --destination-configuration cloudWatchLogs={logGroupName=ivs-demo-logging-config-log-group} Associating a Logging Configuration With an Existing Amazon IVS Chat Room via the Amazon IVS Console We can now associate the logging configuration with an existing Amazon IVS chat room. Select under in the left sidebar, choose the room that you'd like to associate the configuration with, and then click . Rooms Chat Edit On the room edit page, scroll down and select (#1). In the dropdown (#2), search for and associate the configuration that we created above. Note that you can associate multiple configurations with a room, so if you'd also like to log to an Amazon Kinesis Data Firehose or Amazon S3, you can create separate configurations and associate them with your room. Also note that you can create a brand new configuration directly from the edit room page by clicking on the button (#3). Automatically log messages and events Logging configurations Create logging configuration Associating a Logging Configuration With an Existing Amazon IVS Chat Room via the AWS SDK In production, you'll likely use the SDK to associate your logging configuration with a new or existing chat room. To do that, use ( ) or ( ). Here's an example of updating an existing chat room with a chat logging configuration. As expected, we'll need to pass the ARN as the of the chat room, and any/all chat logging configuration ARNs in an array to . CreateRoomCommand docs UpdateRoomCommand docs identifier loggingConfigurationIdentifiers import { IvschatClient, UpdateRoomCommand } from "@aws-sdk/client-ivschat";

const ivsChatClient = new IvschatClient();
const updateChatRoomInput = {
  identifier: 'arn:aws:ivschat:us-east-1:[redacted]:room/[redacted]',
  loggingConfigurationIdentifiers: [
    'arn:aws:ivschat:us-east-1:v:logging-configuration/[redacted]',
  ]
};
const updateChatRoomRequest = new UpdateRoomCommand(updateChatRoomInput);
const updateChatRoomResponse = await ivsChatClient.send(updateChatRoomRequest);
console.log(updateChatRoomResponse); You can specify multiple logging configurations for a chat room. This lets you use multiple destinations, if desired. The array of chat logging identifiers that you pass via the SDK to will overwrite any existing identifiers, so if you are adding an additional configuration, make sure to include any existing configuration ARNs to avoid removing them. Note: UpdateRoomCommand Retrieving Chat Logs At this point, all new messages posted to an Amazon IVS chat room will be logged to the destination(s) specified by the attached logging configuration(s). As mentioned above, in the case of CloudWatch as a logging destionation, we can use the CloudWatch SDK to retrieve our chat logs for a given time period. To do this, we need the , the start and end timestamp (in Unix timestamp format), and the . The will be in the format: . So, for a chat room with an ARN of , the will be . logGroupName logStreamNames logStreamName aws/IVSChatLogs/1.0/room_[suffix of chat room ARN] arn:aws:ivschat:us-east-1:[redacted]:room/0wgOPVl4ZRdu logStreamName aws/IVSChatLogs/1.0/room_0wgOPVl4ZRdu Retrieving CloudWatch Logged Chat Messages Here's how to use the to construct a ( ) that is used to retrieve chat logs. CloudWatchLogsClient FilterLogEventsCommand docs import { CloudWatchLogsClient, FilterLogEventsCommand } from "@aws-sdk/client-cloudwatch-logs";

const cloudwatchClient = new CloudWatchLogsClient();

const filterLogEventsInput = {
  logGroupName: 'demo-chat-log-cw-group',
  logStreamNames: ['aws/IVSChatLogs/1.0/room_0wgOPVl4ZRdu'],
  startTime: 1672929210000,
  endTime: 1672929330000,
};
const filterLogEventsRequest = new FilterLogEventsCommand(filterLogEventsInput);
const filterLogEventsResponse = await cloudwatchClient.send(filterLogEventsRequest);
const events = filterLogEventsResponse.events.map(e => JSON.parse(e.message)); Remember that when streaming to a channel. You might use a stream start and end time if you're trying to retrieve chat logs for a specific Amazon IVS live stream. How do I know which start and end time to use? Amazon IVS sends events to EventBridge Keep in mind, that the can not return an infinite number of events. You may have to check the for the existence of a and handle pagination if your time period has more events than can fit in a single response from the SDK. Refer to the SDK for more information. FilterLogEventsCommand FilterLogEventsCommandOutput nextToken docs By default, this operation returns as many log events as can fit in 1 MB (up to 10,000 log events) or all the events found within the specified time range. If the results include a token, that means there are more log events available. You can get additional results by specifying the token in a subsequent call. This operation can return empty results while there are more log events available through the token. At this point we can do whatever we need with the chat log. Remember, chat logging will log posted to an Amazon IVS chat room, including custom events and events related to chat moderation. If we wanted to replay a chat stream, we'll probably need to parse the events a bit to clean up moderated messages and construct a "replayable" stream of messages. all events : If you haven't read them yet, check out my blog posts on and chat moderation with Amazon IVS chat. More About Moderating Chat automated manual A function to parse the logged events might look similar to this. Your mileage may vary! const parseEvents = (events) => {
  let parsedEvents = [];
  events.forEach(e => {
    switch (e.type) {
      case 'MESSAGE':
        parsedEvents.push(e);
        break;
      case 'EVENT':
        if (e.payload.EventName === 'aws:DELETE_MESSAGE') {
          const existingEventIdx = parsedEvents.findIndex(parsedEvent => {
            return parsedEvent.payload.Id === e.payload.Attributes.MessageID
          });
          if (existingEventIdx > -1) {
            parsedEvents.splice(existingEventIdx, 1);
          }
        }
        break;
    }
  });
  return parsedEvents;
}; This function will return an array that is suitable for chat replay purposes with moderated messages removed from the chat log. Here's a small example of how the event stream might look. [
  {
    "event_timestamp": "2023-01-05T14:33:32.894Z",
    "type": "MESSAGE",
    "payload": {
      "Type": "MESSAGE",
      "Id": "WhO6MW6iRdS5",
      "RequestId": "",
      "Attributes": {
        "username": "gleningp"
      },
      "Content": "bbiab!",
      "SendTime": "2023-01-05T14:33:32.894089757Z",
      "Sender": {
        "UserId": "75758272-f3f2-4f65-83c5-9b8f144116b8",
        "Attributes": {}
      }
    },
    "version": "1.0"
  },
  {
    "event_timestamp": "2023-01-05T14:33:39.896Z",
    "type": "MESSAGE",
    "payload": {
      "Type": "MESSAGE",
      "Id": "VwEPwPV74GN3",
      "RequestId": "",
      "Attributes": {
        "username": "rpeirazzia"
      },
      "Content": "perfect",
      "SendTime": "2023-01-05T14:33:39.896519733Z",
      "Sender": {
        "UserId": "704D6BF8-22D2-4A52-B6A7-BDEFB115ECE5",
        "Attributes": {}
      }
    },
    "version": "1.0"
  },
  {
    "event_timestamp": "2023-01-05T14:33:47.330Z",
    "type": "MESSAGE",
    "payload": {
      "Type": "MESSAGE",
      "Id": "00kqu1sPa6dF",
      "RequestId": "",
      "Attributes": {
        "username": "jmycroft2"
      },
      "Content": "🔥🔥🔥",
      "SendTime": "2023-01-05T14:33:47.330983449Z",
      "Sender": {
        "UserId": "2F12AA2D-DF65-42CF-AE99-133A5A06F7B4",
        "Attributes": {}
      }
    },
    "version": "1.0"
  }
] Summary In this post, we learned how to log messages posted to an Amazon IVS chat room. In our next post, we'll look at bringing together auto-recorded live streams and logged chat messages in an application to create a suitable "on-demand" replay user experience.

How to Archive Web Chat Messages with Amazon IVS Chat Logging

Cómo archivar mensajes de chat web con Amazon IVS Chat Logging

Comment archiver des messages de chat Web avec Amazon IVS Chat Logging

Amazon IVS चैट लॉगिंग के साथ वेब चैट संदेशों को कैसे संग्रहित करें

Amazon IVS チャット ログを使用して Web チャット メッセージをアーカイブする方法

Como arquivar mensagens de bate-papo da Web com registro de bate-papo do Amazon IVS

Cách lưu trữ tin nhắn trò chuyện trên web với Amazon IVS Chat Logging

如何使用 Amazon IVS 聊天记录存档网络聊天消息

The 10 Best Node.js Logging Libraries

Las 10 mejores bibliotecas de registro de Node.js

Les 10 meilleures bibliothèques de journalisation Node.js

10 सर्वश्रेष्ठ नोड.जेएस लॉगिंग लाइब्रेरी

10 の最高の Node.js ロギング ライブラリ

As 10 melhores bibliotecas de registro Node.js

10 thư viện ghi nhật ký Node.js tốt nhất

10 个最好的 Node.js 日志库

Welcome to other chapters of Let’s Understand Chrome V8 Dispatch is responsible for scheduling bytecode, which is equivalent to registering EIP++, and jumps to the next bytecode. Dispatch consists of two parts, one is a dispatch table and the other is the physical register. The table is an array that contains all bytecode addresses. V8 uses the physical register to dispatch bytecode for getting greater efficiency. 1. Dispatch Table The dispatch table is an array of pointers, which is used to store the address of the bytecode processing program. The array type is Code, and the Code Class is below: class Code: public HeapObject {
    public: NEVER_READ_ONLY_SPACE
    using Flags = uint32_t;
    #define CODE_KIND_LIST(V)\
    V(OPTIMIZED_FUNCTION)\
    V(BYTECODE_HANDLER)\
    V(STUB)\
    V(BUILTIN)\
    V(REGEXP)\
    V(WASM_FUNCTION)\
    V(WASM_TO_CAPI_FUNCTION)\
    V(WASM_TO_JS_FUNCTION)\
    V(JS_TO_WASM_FUNCTION)\
    V(JS_TO_JS_FUNCTION)\
    V(WASM_INTERPRETER_ENTRY)\
    V(C_WASM_ENTRY)
    enum Kind {
      #define DEFINE_CODE_KIND_ENUM(name) name,
        CODE_KIND_LIST(DEFINE_CODE_KIND_ENUM)
      #undef DEFINE_CODE_KIND_ENUM
      NUMBER_OF_KINDS
    };
    static
    const char * Kind2String(Kind kind);
    #ifdef ENABLE_DISASSEMBLER
    const char * GetName(Isolate * isolate) const;
    V8_EXPORT_PRIVATE void Disassemble(const char * name, std::ostream & os,
      Address current_pc = kNullAddress);
    #endif
    //................omit................
  }; : In the isolate, there is another pointer array that is the same type as the dispatch table. It holds all the Builtins addresses, which can confuse beginners easily. Note The dispatch table is maintained by BuildWithMacroAssembler, code is below: Code BuildWithMacroAssembler(Isolate * isolate, int32_t builtin_index,
    MacroAssemblerGenerator generator,
    const char * s_name) {
    HandleScope scope(isolate);
    // Canonicalize handles, so that we can share constant pool entries pointing
    // to code targets without dereferencing their handles.
    CanonicalHandleScope canonical(isolate);
    constexpr int kBufferSize = 32 * KB;
    byte buffer[kBufferSize];
    MacroAssembler masm(isolate, BuiltinAssemblerOptions(isolate, builtin_index),
      CodeObjectRequired::kYes,
      ExternalAssemblerBuffer(buffer, kBufferSize));
    masm.set_builtin_index(builtin_index);
    DCHECK(!masm.has_frame());
    generator( & masm);
    int handler_table_offset = 0;
    // JSEntry builtins are a special case and need to generate a handler table.
    DCHECK_EQ(Builtins::KindOf(Builtins::kJSEntry), Builtins::ASM);
    DCHECK_EQ(Builtins::KindOf(Builtins::kJSConstructEntry), Builtins::ASM);
    DCHECK_EQ(Builtins::KindOf(Builtins::kJSRunMicrotasksEntry), Builtins::ASM);
    if (Builtins::IsJSEntryVariant(builtin_index)) {
      handler_table_offset = HandlerTable::EmitReturnTableStart( & masm);
      HandlerTable::EmitReturnEntry( &
        masm, 0, isolate -> builtins() -> js_entry_handler_offset());
    }
    //.....................................................
    //................omit.............................
  } You can see regarding how to debug the BuildWithMacroAssembler. When the value of the parameter builtin_index is 65, the function pointer generator points to which is responsible for maintaining the Dispatch table, code is below: Chapter 9 Generate_InterpreterEnterBytecodeDispatch static void Generate_InterpreterEnterBytecode(MacroAssembler * masm) {
    // Set the return address to the correct point in the interpreter entry
    // trampoline.
    Label builtin_trampoline, trampoline_loaded;
    Smi interpreter_entry_return_pc_offset(
      masm -> isolate() -> heap() -> interpreter_entry_return_pc_offset());
    DCHECK_NE(interpreter_entry_return_pc_offset, Smi::kZero);
    // If the SFI function_data is an InterpreterData, the function will have a
    // custom copy of the interpreter entry trampoline for profiling. If so,
    // get the custom trampoline, otherwise grab the entry address of the global
    // trampoline.
    __ movq(rbx, Operand(rbp, StandardFrameConstants::kFunctionOffset));
    __ LoadTaggedPointerField(
      rbx, FieldOperand(rbx, JSFunction::kSharedFunctionInfoOffset));
    __ LoadTaggedPointerField(
      rbx, FieldOperand(rbx, SharedFunctionInfo::kFunctionDataOffset));
    __ CmpObjectType(rbx, INTERPRETER_DATA_TYPE, kScratchRegister);
    __ j(not_equal, & builtin_trampoline, Label::kNear);
    __ movq(rbx,
      FieldOperand(rbx, InterpreterData::kInterpreterTrampolineOffset));
    __ addq(rbx, Immediate(Code::kHeaderSize - kHeapObjectTag));
    __ jmp( & trampoline_loaded, Label::kNear);
    __ bind( & builtin_trampoline);
    // TODO(jgruber): Replace this by a lookup in the builtin entry table.
    __ movq(rbx,
      __ ExternalReferenceAsOperand(
        ExternalReference::
        address_of_interpreter_entry_trampoline_instruction_start(
          masm -> isolate()),
        kScratchRegister));
    __ bind( & trampoline_loaded);
    __ addq(rbx, Immediate(interpreter_entry_return_pc_offset.value()));
    __ Push(rbx);
    // Initialize dispatch table register.
    __ Move(
      kInterpreterDispatchTableRegister,
      ExternalReference::interpreter_dispatch_table_address(masm -> isolate()));
    // Get the bytecode array pointer from the frame.
    __ movq(kInterpreterBytecodeArrayRegister,
      Operand(rbp, InterpreterFrameConstants::kBytecodeArrayFromFp));
    if (FLAG_debug_code) {
      // Check function data field is actually a BytecodeArray object.
      __ AssertNotSmi(kInterpreterBytecodeArrayRegister);
      __ CmpObjectType(kInterpreterBytecodeArrayRegister, BYTECODE_ARRAY_TYPE,
        rbx);
      __ Assert(
        equal,
        AbortReason::kFunctionDataShouldBeBytecodeArrayOnInterpreterEntry);
    }
    // Get the target bytecode offset from the frame.
    __ movq(kInterpreterBytecodeOffsetRegister,
      Operand(rbp, InterpreterFrameConstants::kBytecodeOffsetFromFp));
    __ SmiUntag(kInterpreterBytecodeOffsetRegister,
      kInterpreterBytecodeOffsetRegister);
    // Dispatch to the target bytecode.
    __ movzxbq(r11, Operand(kInterpreterBytecodeArrayRegister,
      kInterpreterBytecodeOffsetRegister, times_1, 0));
    __ movq(kJavaScriptCallCodeStartRegister,
      Operand(kInterpreterDispatchTableRegister, r11,
        times_system_pointer_size, 0));
    __ jmp(kJavaScriptCallCodeStartRegister);
  }
  //================================separation=========================
  void Builtins::Generate_InterpreterEnterBytecodeDispatch(MacroAssembler * masm) {
    Generate_InterpreterEnterBytecode(masm);
  } In the above code, the is the entry point of bytecode dispatch, and the is the key code that implements the bytecode dispatch. Here are two important concepts: Generate_InterpreterEnterBytecodeDispatch Generate_InterpreterEnterBytecode The is a Builtin and its index is 65. (The index may be dissimilar in different V8 versions) Generate_InterpreterEnterBytecodeDispatch V8 uses a physical register to manage the dispatch table, the name of the register is . Using the physical register can avoid the push and pop of the dispatch table frequently, to improve efficiency. kInterpreterDispatchTableRegister Here are the key lines of code： Line 35, loads the dispatch table to the A register. Line 37 of code, take out the table address from the isolate. The implementation of A is in the below three functions. (1) ExternalReference ExternalReference::interpreter_dispatch_table_address(
  Isolate * isolate) {
  return ExternalReference(isolate -> interpreter() -> dispatch_table_address());
}
interpreter::Interpreter * interpreter() const {
  return interpreter_;
}
Address dispatch_table_address() {
  return reinterpret_cast < Address > ( & dispatch_table_[0]);
} Line 39 takes out the bytecode array from the stack. (2) Line 51 takes out the offset of the target bytecode from the bytecode array. (3) Lines 56 and 58 calculate the target bytecode address and store the address in kJavaScriptCallCodeStartRegister. (4) Line 61 jumps to register kJavaScriptCallCodeStartRegister and executes bytecode. Figure 1 is the call stack. (5) 2. TailCall The Dispatch() is the last step in a bytecode, so its alias is the tail call. Let’s look at a few bytecodes： IGNITION_HANDLER(StaGlobal, InterpreterAssembler) {
    TNode < Context > context = GetContext();
    //omit..............
    Goto( & end);
    Bind( & no_feedback);
    CallRuntime(Runtime::kStoreGlobalICNoFeedback_Miss, context, value, name);
    Goto( & end);
    Bind( & end);
    Dispatch(); // !!! this is the tail call.
  }
  IGNITION_HANDLER(LdaContextSlot, InterpreterAssembler) {
    TNode < Context > context = CAST(LoadRegisterAtOperandIndex(0));
    TNode < IntPtrT > slot_index = Signed(BytecodeOperandIdx(1));
    TNode < Uint32T > depth = BytecodeOperandUImm(2);
    TNode < Context > slot_context = GetContextAtDepth(context, depth);
    TNode < Object > result = LoadContextElement(slot_context, slot_index);
    SetAccumulator(result);
    Dispatch(); //!!! here also is.
  }
  IGNITION_HANDLER(LdaImmutableContextSlot, InterpreterAssembler) {
    TNode < Context > context = CAST(LoadRegisterAtOperandIndex(0));
    TNode < IntPtrT > slot_index = Signed(BytecodeOperandIdx(1));
    TNode < Uint32T > depth = BytecodeOperandUImm(2);
    TNode < Context > slot_context = GetContextAtDepth(context, depth);
    TNode < Object > result = LoadContextElement(slot_context, slot_index);
    SetAccumulator(result);
    Dispatch(); //!!! here 
  } The above three bytecodes all end with Dispatch(). Below is the code of Dispatch: void InterpreterAssembler::Dispatch() {
    Comment("========= Dispatch");
    DCHECK_IMPLIES(Bytecodes::MakesCallAlongCriticalPath(bytecode_), made_call_);
    TNode < IntPtrT > target_offset = Advance();
    TNode < WordT > target_bytecode = LoadBytecode(target_offset);
    if (Bytecodes::IsStarLookahead(bytecode_, operand_scale_)) {
      target_bytecode = StarDispatchLookahead(target_bytecode);
    }
    DispatchToBytecode(target_bytecode, BytecodeOffset());
  } The 2nd line is the comment function. When debugging, we can use its output information to locate the execution position. The 4th line takes out the offset of the target(next) bytecode. The 5th line loads the target bytecode. The 9th line calls which calls eventually. DspatchToBytecode() DispatchToBytecodeHandlerEntry() void InterpreterAssembler::DispatchToBytecodeHandlerEntry(
  TNode < RawPtrT > handler_entry, TNode < IntPtrT > bytecode_offset) {
  // Propagate speculation poisoning.
  TNode < RawPtrT > poisoned_handler_entry =
    UncheckedCast < RawPtrT > (WordPoisonOnSpeculation(handler_entry));
  TailCallBytecodeDispatch(InterpreterDispatchDescriptor {},
    poisoned_handler_entry, GetAccumulatorUnchecked(),
    bytecode_offset, BytecodeArrayTaggedPointer(),
    DispatchTablePointer());
} In , parameter 1 is the target bytecode and 2 is the offset of the bytecode. Parameter 1 is used to load the bytecode handler and 2 is used to load operands. Then call , which is below: DispatchToBytecodeHandlerEntry TailCallBytecodeDispatch template < class...TArgs >
    void CodeAssembler::TailCallBytecodeDispatch(
      const CallInterfaceDescriptor & descriptor, TNode < RawPtrT > target,
        TArgs...args) {
      DCHECK_EQ(descriptor.GetParameterCount(), sizeof...(args));
      auto call_descriptor = Linkage::GetBytecodeDispatchCallDescriptor(
        zone(), descriptor, descriptor.GetStackParameterCount());
      Node * nodes[] = {
        target,
        args...
      };
      CHECK_EQ(descriptor.GetParameterCount() + 1, arraysize(nodes));
      raw_assembler() -> TailCallN(call_descriptor, arraysize(nodes), nodes);
    } In the above code, the 5th line creates the descriptor, as below: CallDescriptor * Linkage::GetBytecodeDispatchCallDescriptor(
    Zone * zone,
    const CallInterfaceDescriptor & descriptor,
      int stack_parameter_count) {
    const int register_parameter_count = descriptor.GetRegisterParameterCount();
    const int parameter_count = register_parameter_count + stack_parameter_count;
    DCHECK_EQ(descriptor.GetReturnCount(), 1);
    LocationSignature::Builder locations(zone, 1, parameter_count);
    locations.AddReturn(regloc(kReturnRegister0, descriptor.GetReturnType(0)));
    for (int i = 0; i < parameter_count; i++) {
      if (i < register_parameter_count) {
        // The first parameters go in registers.
        Register reg = descriptor.GetRegisterParameter(i);
        MachineType type = descriptor.GetParameterType(i);
        locations.AddParam(regloc(reg, type));
      } else {
        int stack_slot = i - register_parameter_count - stack_parameter_count;
        locations.AddParam(LinkageLocation::ForCallerFrameSlot(
          stack_slot, MachineType::AnyTagged()));
      }
    }
    MachineType target_type = MachineType::Pointer();
    LinkageLocation target_loc = LinkageLocation::ForAnyRegister(target_type);
    const CallDescriptor::Flags kFlags =
      CallDescriptor::kCanUseRoots | CallDescriptor::kFixedTargetRegister;
    return new(zone) CallDescriptor( // --
      CallDescriptor::kCallAddress, // kind
      target_type, // target MachineType
      target_loc, // target location
      locations.Build(), // location_sig
      stack_parameter_count, // stack_parameter_count
      Operator::kNoProperties, // properties
      kNoCalleeSaved, // callee-saved registers
      kNoCalleeSaved, // callee-saved fp
      kFlags, // flags
      descriptor.DebugName());
  } In , parameter 2 is the target bytecode, and parameter 3 is the stack parameter count. This function is used to apply for register resources and create a . The gives what parameters to supply when executing a bytecode. The comments on lines 24 to 25 give its layout. GetBytecodeDispatchCallDescriptor CallDescriptor CallDescriptor In , the on the 9th line is below, we see that it’s parameter 1 is . TailCallBytecodeDispatch TailCallN() CallDescriptor void RawMachineAssembler::TailCallN(CallDescriptor * call_descriptor,
    int input_count, Node *
    const * inputs) {
    // +1 is for target.
    DCHECK_EQ(input_count, call_descriptor -> ParameterCount() + 1);
    Node * tail_call =
      MakeNode(common() -> TailCall(call_descriptor), input_count, inputs);
    schedule() -> AddTailCall(CurrentBlock(), tail_call);
    current_block_ = nullptr;
  } The 5th line generates a node. In the 7th line adds the node to the end of the current basic block. So far, the dispatch is completed and the next bytecode is started to execute. Figure 2 shows the function call stack. AddTailCall() Okay, that wraps it up for this share. I’ll see you guys next time, take care! Please reach out to me if you have any issues. : qq9123013 : v8blink@outlook.com WeChat Email Also published . here

Stories, News & Mentions about JOYSTREAM

Joystream’s Stories on HackerNoon