Streamhub is a B2B video analytics platform that helps customers build their best loyal audiences to their videos according to interests

Streamhub

streamhub

Creating and Releasing an App with Flutter: Part II

In this tutorial, we’ll be using Alpaca, Python, Docker, and AWS to create a fully scalable algorithmic crypto trading bot that can execute in your sleep. The strategy that will be used is based on the popular . You can check out all the source code for this article in this or in this . Supertrend indicator GitHub repository Google Colab Without further ado, let’s get started! Step 1: Creating Alpaca and AWS Accounts Setting up Alpaca In order to create the algorithmic trading bot system used in this article, the first step is to create an Alpaca account. If you don’t have an account yet, you can for free! Alpaca offers both a live brokerage as well as a paper trading platform but you can sign up just for paper trading. sign up here After creating an account, you will need to create your API credentials to access Alpaca data via Python. Save both the API Key as well as the Secret Key into a safe place on your computer. These keys will allow us to retrieve live price data for cryptocurrencies as well as execute trades in our paper trading account. Setting up AWS In order to deploy the trading bot, we’ll need an AWS account. You can for a free tier account, which offers more than 100 different services and in our case, 2200 hours of Amazon ECS Anywhere coverage. Fortunately, the sign-up process is relatively straightforward, but to learn more about what the free tier consists of, check out this . sign up link Step 2: Creating Trading Bot in Python Now that we have the accounts necessary to get started, we can start building the trading bot in Python. For this tutorial, we’ll be creating a bot that trades Bitcoin based on the signals of the Supertrend indicator. The Supertrend indicator is a popular tool for crypto swing trading as well as trend following systems but can be adapted to any timeframe. It’s calculated by using the difference between the current average true range and the average price from the most recent price bar. Import Dependencies First, we’ll need to import all of the required dependencies we’ll be using in the bot including smtplib for email communication, pandas-ta for the calculation of the supertrend indicator, and alpaca-trade-api for live market data and paper trading account access. If you have not yet installed some of these dependencies, you can go ahead and pip install them in your terminal. # Import Dependencies
import smtplib
import pandas as pd
import pandas_ta as ta
import alpaca_trade_api as tradeapi
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart Define Variables The next step is to define most of the general variables we’ll be needing throughout the program. For the API and Secret keys, you can paste the values you stored from step 1 of this tutorial when creating an Alpaca account. For the trading strategy, we’ll be buying and selling Bitcoin and each transaction will consist of 1 quantity. # API Credentials
API_KEY='********'
SECRET_KEY='********'
api = tradeapi.REST(API_KEY, SECRET_KEY,'https://paper-api.alpaca.markets')

# Define crypto related variables
symbol = 'BTCUSD'
qty_per_trade = 1 Create Function to Check Account Positions Next, we have to create a function to check whether the trading account currently holds any Bitcoin. If it does, we can return the quantity of Bitcoin that is being held. If there is no Bitcoin, then the function will just return 0. This is important because, in the next function which handles the buying and selling, we can focus on buying only if there is currently no Bitcoin in the account. Otherwise, we’ll monitor the sell signal to see if the position should be closed. # Check Whether Account Currently Holds Symbol
def check_positions(symbol):
    positions = api.list_positions()
    for p in positions:
        if p.symbol == symbol:
            return float(p.qty)
    return 0 Create Function to Trade Bitcoin Finally, we can create a function to retrieve the data, calculate the supertrend indicator signals, and execute the buy/sell orders. First, the function takes in an input of bar data which is used for the live current close price. Using the pandas-ta module discussed earlier, we can calculate the supertrend indicator as a column and concatenate it to the dataframe with pricing information. If the closing price is above the supertrend indicator, then a buy signal is issued. If the closing price is below the supertrend indicator, then a sell signal is issued. So by using the latest values from the dataframe, we can create boolean values for buy and sell signals. We can set the code to buy 1 quantity of Bitcoin if there is currently no position and if a buy signal is true. Likewise, we can set the code to sell 1 quantity of Bitcoin if there is currently a position and if a sell signal is true. Wrapping the entire function in a try/except block ensures that the program will not break due to errors, and will simply print out the error message. Since this is a trading bot and is intended to run throughout market hours, it’s best if the program is continuously running. # Supertrend Indicator Bot Function
def supertrend_bot(bar):
    try:
        # Get the Latest Data
        dataframe = api.get_crypto_bars(symbol, tradeapi.TimeFrame(1, tradeapi.TimeFrameUnit.Minute)).df
        dataframe = dataframe[dataframe.exchange == 'CBSE']
        sti = ta.supertrend(dataframe['high'], dataframe['low'], dataframe['close'], 7, 3)
        dataframe = pd.concat([dataframe, sti], axis=1)

        position = check_positions(symbol=symbol)
        should_buy = bar["c"] > dataframe["SUPERT_7_3.0"][-1]
        should_sell = bar["c"] < dataframe["SUPERT_7_3.0"][-1]
        print(f"Price: {bar["c"]}")
        print("Super Trend Indicator: {}".format(dataframe["SUPERT_7_3.0"][-1]))
        print(f"Position: {position} | Should Buy: {should_buy}")

        # Check if No Position and Buy Signal is True
        if position == 0 and should_buy == True:
            api.submit_order(symbol, qty=qty_per_trade, side='buy')
            message = f'Symbol: {symbol} | Side: Buy | Quantity: {qty_per_trade}'
            print(message)
            # send_mail(message)

        # Check if Long Position and Sell Signal is True
        elif position > 0 and should_sell == True:
            api.submit_order(symbol, qty=qty_per_trade, side='sell')
            message = f'Symbol: {symbol} | Side: Sell | Quantity: {qty_per_trade}'
            print(message)
            # send_mail(message)
        print("-"*20)

    except Exception as e:
        print (e)
        # send_mail(f"Trading Bot failed due to {e}") Step 3: Implementing Email Communication System Create Function for Sending Emails A critical aspect of any algorithmic trading system is having a built-in communication system where updates can be sent about any potential errors, transactions, or status. With Python, it’s extremely easy to set up email messaging using the packages email and smtplib. The first step is to set the variables for the sending and receiving emails. The email that is sending the messages must have the setting set to be off. Due to security reasons, it’s for the best if you create an entirely new email just for this trading strategy. “Less Secure Apps” Next, we can set the “from” tag equal to the algorithm name, subject title equal to our bot, and mail content using the input of text for the function. Using smtplib, we can access the gmail server, log in using the credentials given, and send the email message. You can now uncomment all of the send_mail function calls from the previous step! # Send an Update Email After Every Trade
def send_mail(message_text):
    # Define email related variables
    sender_email = '**************'
    sender_password = '**************'
    receiver_email = '**************'

    # Create a multipart message and set headers
    message = MIMEMultipart()
    message['From'] = 'Crypto Trading Algorithm'
    message['To'] = receiver_email
    message['Subject'] = 'Supertrend Indicator Bot'

    # Add body to email and send
    mail_content = message_text
    message.attach(MIMEText(mail_content, 'plain'))
    session = smtplib.SMTP('smtp.gmail.com', 587)
    session.starttls()
    session.login(sender_email, sender_password)
    text = message.as_string()
    session.sendmail(sender_email, receiver_email, text)
    session.quit()
    print ('Mail Sent')

    return {"Success": True} Step 4: Stream Live Data From Alpaca Set Up Alpaca Live Crypto Data The last step of building the Python bot is to start streaming live market data for Bitcoin from Alpaca. Fortunately, Alpaca makes this process extremely easy. First, we have to create an instance of the data streaming API by calling the Stream method in which we pass the API keys. We can also specify that we want raw data only from the Coinbase exchange. Then, we can create an asynchronous function to receive the live bar data and within this function, we can call the supertrend_bot function we made in step 2. Here, we can send an email that the trading bot is now live, subscribe to crypto data, and start streaming live data! # Create instance of Alpaca data streaming API
alpaca_stream = tradeapi.Stream(API_KEY, SECRET_KEY, raw_data=True, crypto_exchanges=['CBSE'])

# Create handler for receiving live bar data
async def on_crypto_bar(bar):
    print(bar)
    supertrend_bot(bar)

send_mail("Trading Bot is Live")

# Subscribe to data and assign handler
alpaca_stream.subscribe_crypto_bars(on_crypto_bar, symbol)

# Start streaming of data
alpaca_stream.run() Step 5: Dockerizing the Python Program Creating a Dockerfile In the next step, we’ll be deploying the code to AWS Elastic Container Services. To do that, we must first dockerize the python program we have just created. If you have not already done so, the first step is to onto your machine. Once you’ve done so, you should be able to run the following command in your terminal and receive a valid response such as “Docker version 20.10.14, build a224086”. download Docker docker -v Next, you must enter into your project directory in which you have the python program we have created and then create a Dockerfile. Using the terminal, you can simply run the command below! touch Dockerfile So far, the project directory should be something similar to what’s below. .
├── crypto_bot.py
├── Dockerfile Now, we can start populating the Dockerfile with the information it needs to containerize the application. First, we should pull down a tagged Python base image using the FROM command. The version 3.7.6 can be changed to suit your needs. Then, we can ADD the source file to the container’s base folder which in this case is crypto_bot.py. In order to use third-party libraries like alpaca-trade-api and pandas-ta in our docker app, we have to install them using the RUN pip install command. Lastly, we can add the final command that Docker will run once it’s fully set up. At this point, our Dockerfile is complete and we just have to build it within our terminal. FROM python:3.7.6

ADD crypto_bot.py .

RUN pip install alpaca-trade-api pandas-ta

CMD [ "python", "./crypto_bot.py" ] In order to build the docker app, we can return to our terminal in the same project directory and run the following command. You can replace “python-imagename” with whatever name you’d like to call this docker app! docker build -t python-imagename . In order to build the docker app, we can return to our terminal in the same project directory and run the following command. You can replace “python-imagename” with whatever name you’d like to call this docker image! Once the docker app is done building, you can run the following command to run it and you should see that the Python program will run as usual. Hopefully, you should have received an email that the trading bot is live and that trades can be executed. Keep in mind that with any changes you make to the program, you must rebuild the docker app using the previous command and run it again with the command below. And that’s it, our Python program has been fully dockerized! docker run -t python-imagename Step 6: Deploying to AWS Cloud Provider AWS CLI Configuration In order to get our docker app up and running in the cloud, we first need to if it has not already been installed. download the AWS command-line interface Once the CLI is installed, we can run the following command in our terminal to log in to our AWS account and get started with the deployment. aws configure After running this command, your terminal will ask you to provide your AWS Access Key ID and your AWS Secret Access Keys. Both of these can be found within your AWS account within security credentials. You can leave the default region name and output format as they currently are. AWS ECS Repositories Now that we’ve logged into our account in the terminal, the next step is to get our local docker app into an ECS repository in AWS. To do that, type in “ECS Repositories” in the search bar and click on the first Features link. On the ECS repositories screen, you can go ahead and create a new repository, give it a name, and leave the rest of the settings as their default state. Now, you should have a screen that’s similar to the one below. If you open up the repository and click on the “View push commands” button, you’ll be able to see the commands that are needed to push our local docker app into this repository. Follow the instructions on your screen (similar to the one below) in your local terminal. Your app should now be pushed to AWS! AWS ECS Clusters In the AWS console, navigate to AWS ECS and click the “create cluster” button. For the cluster template, click on “EC2 Linux + Networking” and continue to the next step. On the next page, you can configure the cluster to suit your specific needs. In this particular case, you only need to change six things. Create a name for your cluster Under instance configuration, change the EC2 instance type to t3a.micro Under networking, select the default dropdown value for VPC Under networking, select the default dropdown value for Subnets Under networking, select Enabled for “Auto assign public IP” Under networking, select the default dropdown value for “Security Group” After these steps, you can go ahead and create the cluster. Once, the process has been completed you should see the ECS cluster has been created, the ECS Instance IAM Policy was attached, and the CloudFormation Stack was created. AWS Task Definitions Once your cluster has been created, click on the “Task Definitions” button under Clusters in the left-hand sidebar. Click on create a new task definition, pick EC2 as the launch type compatibility and continue on to the next step. Enter a task name and scroll down to the section named Task size. Here, you can enter whatever memory and CPU values you are comfortable with. Then, click on Add container under “Container definitions.” Enter a name once again and for the image, paste the container image URL that you received from AWS. It should be similar to “https://public.ecr.aws/g9l6a0m7/crypto-bot:latest.” Then, you can scroll to the bottom of the page and create the task definition. AWS Deploying Tasks Finally, return back to the main clusters page so that you can see a screen similar to the one below. Enter into your cluster, and navigate to the Tasks tab. There, you should see an option to run a new task. Click on that button and a new popup wIll appear like the one below. Select EC2 as the launch type, and under task definition, select the name of the task you created in the previous step. Enter in the name of the cluster you just created under the Cluster dropdown. Then, go ahead and run the task. At this point, the app should be up and running on AWS! Conclusion In this tutorial, we covered everything from creating a live trading bot and implementing email communication to dockerizing a Python program and deploying it onto AWS ECS. While the strategy in the article was basic and not intended to be used as a full trading system, the processes to build a fully scalable system around a Python bot are the same so you can now automate anything you’d like! Thanks for reading, and I hope you learned something about creating algorithmic trading bots with the Alpaca Crypto API! Please note that this article is for general informational purposes only. All screenshots are for illustrative purposes only. Alpaca does not recommend any specific cryptocurrencies. Cryptocurrency services are made available by Alpaca Crypto LLC ("Alpaca Crypto"), a FinCEN registered money services business (NMLS # 2160858), and a wholly-owned subsidiary of AlpacaDB, Inc. Alpaca Crypto is not a member of SIPC or FINRA. Cryptocurrencies are not stocks and your cryptocurrency investments are not protected by either FDIC or SIPC. Depending on your location, cryptocurrency services may be provided by West Realm Shires Services, Inc., d/b/a FTX US (NMLS #1957771). This is not an offer, solicitation of an offer, or advice to buy or sell cryptocurrencies, or open a cryptocurrency account in any jurisdiction where Alpaca Crypto, or FTX US respectively, are not registered or licensed, as applicable. Cryptocurrency is highly speculative in nature, involves a high degree of risks, such as volatile market price swings, market manipulation, flash crashes, and cybersecurity risks. Cryptocurrency is not regulated or is lightly regulated in most countries. Cryptocurrency trading can lead to large, immediate and permanent loss of financial value. You should have appropriate knowledge and experience before engaging in cryptocurrency trading. For additional information, please click . here Please see alpaca.markets and Alpaca’s Disclosure Library for more information.

How to Build an Algorithmic Crypto Trading Bot in 6 Steps

What you need to know about writing smart contracts on the Tezos blockchain when you come from Ethereum The Tezos blockchain is one of the main projects you need on your radar for 2020. Its liquid proof of stake system makes it increasingly attractive for people looking for steady revenues. Its token has more than doubled in price since last December, making it interesting for investors. And there are many new projects planned to be deployed on the blockchain in 2020 (an , , and the new to name a few). NFT-based game USDTez tzBTC You have some experience with the Ethereum blockchain and you’re thinking about taking the plunge and getting your feet wet with Ligo. Congratulations! However, you’ll quickly notice that even if they’re both labeled “blockchain,” programming on the Tezos blockchain and the Ethereum blockchain can be as different as programming in two different languages. Some concepts that you know from Solidity do not apply (and can even be counterproductive) when coding with Ligo. Likewise, some concepts in Ligo do not exist in Solidity. In this article, we’ll look at the main differences, pitfalls, and things to keep in mind when you are used to Solidity and want to join us and become a Tezos smart contract developer. I’ll assume that you have a (moderately) good knowledge of Solidity and JavaScript. No knowledge of Ligo is required, although it may help you to better understand the comparisons. The Type System The type system for simple values in Solidity and Ligo are very similar. Ligo has integers, strings, addresses, nat (the unsigned integer of Solidity), booleans, etc. One of the perks of types of Ligo is that you can declare your own types, it allows for a better structure and organization of your code. For example, it may be clearer to sometimes create a new type instead of using a primitive type ( instead of , instead of ). username string score nat The complex types present more differences: Ligo has tuples, sets, and lists but no arrays. Sets can be seen as unordered arrays, containing unique values of the same type, while lists can be seen as ordered arrays containing values of the same type. Ligo also provides a number of methods to read, modify, or return information about the sets and lists, which are very handy. Ligo also offers which are very useful and similar to in Solidity. Ligo doesn’t have enums, but that give you all the functionalities of enums and additional benefits! A special kind of variant called will spare you a lot of bugs when verifying the right value is provided. records structs variants option Another major difference between Solidity and Ligo is in its maps (the mappings of Solidity), which is the subject of the next paragraph. Maps and Big Maps Ligo offers maps and big maps, which behave essentially in the same way. Big maps are more suitable for containing a large amount of data. They resemble mappings of Solidity with a few benefits. In Solidity, all the keys of a mapping exist by default, if you try to get the value associated with a key that doesn’t exist, you will get the default value of its type. This is not the case in Ligo. Thanks to pattern matching, you know precisely when you query a value in a map whether it returns something or not. If nothing is returned, you get an . You aren’t left wondering whether the or the empty string returned is the actual value or the default value because the key doesn’t exist! option with the value None 0 In Ligo, it’s possible to iterate over a map (this is not possible natively in Solidity with mappings). When iterating over a map, you can just check the returned values, you can modify them or you can “ ” them. One feature that takes some time to get used to when coming from Solidity (or a language like JavaScript) is that the functions you will use are not methods of the instance of the map (or the set/list) but methods of the Map (list/set) object. For example, if you want to know the size of a map you wouldn’t write but . To add an element to a set, you wouldn’t write but . fold my_map.size() Map.size(my_map) my_set.add(el) Set.add(el, my_set) String Manipulation and Sets/lists I put these two different cases together as they are little highlights that will make your job easier and your code safer. Solidity is infamous for lacking any sort of native string manipulation — you can’t slice them, concatenate them, etc. Rejoice, because Ligo lets you do it! You get three methods that will help you work with strings. will return the length of the string, will return a substring of a string and strings can easily be concatenated in ReasonLigo with the operation ( ). String.length() String.sub() ++ string1 ++ string2 I particularly enjoy working with sets and lists when writing smart contracts in Ligo. Sets are unordered lists of unique values of the same type. A few functions help you manipulate them easily ( , ) and you can be sure you have unique non-repeated values in them. Set.add() Set.remove() Lists are ordered collections of elements of the same type. If you need to store some values in a certain order, this is what you use. Different methods help you work with lists, like , and . These types add another level of security and robustness to your smart contract. List.iter() List.map() List.fold() Contract Deployment and Initialization One of the highlights of being a smart contract developer is deploying your contract to the blockchain and letting it start its self-contained life! When you write a smart contract in Solidity, you usually start by building the constructor. This function is in charge of initializing storage during the deployment of the contract. This doesn’t exist in Ligo. There’s no constructor function. Instead, you provide the initial storage while deploying the smart contract! In the Tezos lingo, it is called “ ”. originating a contract Entry Points The logic of using a smart contract on the Tezos blockchain is somewhat different from that of a contract on the Ethereum blockchain. In Solidity, you have multiple access points of your contract that you can call directly to modify or return some data. This creates the need for a “fallback function” in case users try to call a function that doesn’t exist. In Ligo, there’s only one single access point for your smart contract called the entry point. This may sound a little weird at the beginning (it did to me) but it’s actually a robust system that wipes out the need for a fallback function. Your main entry point receives the name of the function you’re trying to call as a parameter and will use pattern matching to ensure that the function exists. Then, it will call the matching function called a “pseudo-entry point.” If it doesn’t exist, it will just fail. In Solidity, calling a function that doesn’t exist triggers the fallback function. In Ligo, it throws an error. Note: The main entry point of your smart contract in Ligo only accepts two parameters: one parameter sent with the transaction and the storage. It’s not possible to provide multiple parameters like in Solidity . However, you can send a record to the function. This restriction is helpful in avoiding common bugs in Solidity when passing parameters in the wrong order. function myFunc(param1 uint, param2 uint, param3 string) State Updates One of the other main differences between Solidity and Ligo is in the transaction flow. In Solidity, when you send a transaction to a smart contract, the state of the smart contract is updated along the way until the transaction reaches the end of the function. If there’s an error at any point in the flow (for example thrown by an or exception), the state is reverted to its original value before the transaction happened. This leaves your code open to s if you are not careful about when and how you update your state. assert require reentrancy attack Reentrancy attacks are unlikely with a smart contract on the Tezos blockchain. In Ligo, the main entry point returns two things: a list of transactions to execute and the new state. At no point during the transaction flow is the storage of your smart contract altered. If there’s an error, the transaction will just stop with nothing changed. Since the function returns the whole storage, pattern matching will make sure that you return correct storage with the expected values. If a property of your storage is supposed to be a nat, it will be impossible to store an integer instead. always Pattern Matching Pattern matching is among the features of a smart contract in Ligo that will help you make it bug-free. Below are two examples of pattern matching, but this is a built-in feature that will prove itself useful all over your smart contract! In the pseudo-entry points As mentioned earlier, there is only one “true” entry point in your Ligo smart contract. Through pattern matching, it will be in charge of redirecting the transaction call to the right function. Let’s check the example provided on the Ligo website: When the transaction reaches the smart contract, it’s evaluated against its name and its parameters. If the type of the parameter matches, it will go through the switch pattern to be redirected to its corresponding function. After the function processes the transaction, the new storage is returned. In updating maps Pattern matching is a powerful ally when you want to update a map. In Solidity, you have to take a wild guess when updating a mapping and assume the key has been initialized with a value (because all keys exist by default). Sometimes it’s a desirable pattern — for example, if you want to create a new key because adding a value to a key that was not previously initialized in a mapping will just create it. In Ligo, it’s impossible to add a value to a key that doesn’t exist in a map. Every search and update of a map goes through pattern matching. Imagine we have a mapping of the type that stores the number of tezzies owned by our users. This is how you would update the value for a provided address: map(address, tez) ( .find_opt (user, wallets)) {
  | Some (wallet) => .update(user, wallet + tez, wallets)
  | failwith ( ) : return_type
} switch Map Map 1 => None "No wallet found" Here’s what goes on in this little piece of code. is a function that returns an , i.e a variant type that contains only two options. Either there’s something ( ) or there is nothing ( ). If a value associated with the key was found, the value is returned with . Otherwise, is returned (in this case, we just throw an error, but you could also create the key/value pair or complete another action). It’s then impossible to try to update the value of a key that doesn’t exist. Map.find_opt option type Some None Some None In Solidity, you would write because by default the key exists and you have no control over its existence. The wallet could also be equal to and you wouldn’t know if it is a desirable value or just the default value. In Ligo, if the value of a key is , you know that it was set to be equal to 0 on purpose. wallets[user] += 1 ether 0ether 0tez Storage Reading One of the most puzzling differences between Ligo and Solidity for me, when I started writing smart contracts for the Tezos blockchain, was how to get the values from the contract storage out to my dapp. As mentioned earlier, the entry points of a Ligo contract always return the whole storage and a list of operations. There is no getter function to implement if you want to return one single value. Also, the storage in your Ligo smart contract is one single variable (generally a record), instead of distinct separate variables as in Solidity. I only got a satisfying answer to this question when I started playing with libraries for dapps, like the excellent . For every smart contract developer coming from Solidity and wondering how Tezos smart contracts export their values to use them in a dapp, the simple answer is: they don’t. They won’t any value but you can them from outside. The Tezos node exposes the contract storage and libraries like Taquito give you access to the whole storage (for example with if using Taquito). Taquito return read await contractInstance.storage() Solidity has to automatically create getter functions for your variables, while this is not a concern with Ligo. In Ligo every value of the storage is always one promise away from you! Desirable Features From Solidity Now that I’ve spent the whole article trying to show you how amazing writing smart contracts for the Tezos blockchain is, let’s see a few things that you’re going to miss as a Solidity developer when switching to Ligo. Obviously, the Tezos ecosystem hasn’t yet reached the maturity of the Ethereum one and a lot of tools and features are yet to be developed. Here are a few that will hopefully make their way into the smart contracts and the Tezos ecosystem: Unlike in Ethereum, there are no events in Tezos smart contracts. You won’t be notified that the state of the smart contract changed. Taquito has a function to listen to storage changes ( ) but I tried it on a sandboxed node and half of the storage updates didn’t register. As of today, the most reliable way to get notified of storage updates is a good old with a function that compares the values of the freshly fetched storage and the old one. Events. Tezos.stream.subscribeOperation setInterval Remix is like the Holy Grail of writing, testing, and deploying smart contracts online. Ligo (and for that matter) have done a great job creating online editors but they still feel flimsy compared to Remix. With the Ligo editor, you can write your smart contracts (the syntax highlighting and autocompletion of variable names and methods/properties are neat), compile them to Michelson, test your entry points and deploy it to Carthagenet (in theory). However, the editor doesn’t let you save your smart contract — a wrong swipe on my Magic Mouse meant that several times I had to start again. It takes a few try-and-fails to understand how to format the initial values of your storage to test the smart contract entry points. Finally, there is no JavaScript VM for testing. Remix. SmartPy Sorry to be the bearer of bad news, but there’s no MetaMask for the Tezos blockchain. You’re left wondering how the users of your dapp can sign their transactions until you find (and accept) TezBridge. TezBridge is a great tool as it allows dapp users to sign their transactions without installing anything and it integrates nicely with Taquito. However, you are going to miss the elegance, simplicity, and convenience of MetaMask. You won’t be able to check the history of your transactions, you won’t see your balance, the tokens you own, and you won’t be able to send tezzies from the interface. TezBridge does just one thing, help you sign your transactions, and it does it well! MetaMask. Because Tezos is only two years old, the available tools to help you develop smart contracts and dapps are still in their infancy and pretty scarce. The ones that are available are generally not beginner-friendly, as you must have a good knowledge of Michelson and/or functional programming to understand documentations that are either incomplete or highly technical. If you’re a beginner developer in JavaScript, it can take a few hours to write a simple smart contract in Solidity, figure out how to test it in Remix, spin up Ganache and Truffle and build a simple dapp with web3js. It won’t happen when you switch to Tezos (at least for the moment). You’ll have to do some research, read documentation, bug the amazing and very helpful people on the Ligo Telegram, and try and fail until you find the right way to do it. The tools in general. There are countless tutorials out there to guide you, whether you want to create a simple smart contract in Solidity, use OpenZeppelin, or create upgradable contracts or develop the next DeFi hit. This is not the case with Tezos. You’ll find some tutorials for simple things but for more advanced concepts tutorials are scarce or nonexistent. Learning Ligo will require you to have a basic understanding of functional programming (although ReasonLigo makes it a lot easier). Many of the smart contracts out there I could find for references are written in PascaLigo, which adds another level of complexity for beginners. The resources. Conclusion As you can imagine, switching from Solidity to Ligo requires patience and a lot of reading. You have to learn how to use a new programming language for smart contracts that’s pretty different from Solidity. You need to find new tools to help you test your smart contracts and develop your dapps. You need to learn how to interact with smart contracts on Tezos from your application. If you are a self-taught developer like me, it will take you some time to figure out how everything works together. However, it’s a very rewarding process. During these past months, I’ve learned a lot about functional programming, smart contracts, Tezos, and dapps in general. If you continue writing smart contracts in Solidity, a glimpse of another approach to creating them may help you become a better developer. Just be cautious — writing smart contracts for Tezos is highly addictive, as soon as you realize how much secure, robust, and elegant they are! Don’t hesitate to leave your opinions or suggestions!

Solidity vs. Ligo: The Key Differences

Retrieval-Augmented Generation (RAG) blends the generative abilities of LLMs with the retrieval of information from diverse knowledge bases (KBs). However, traditional implementations of RAG have often relied on pre-scripting the logic for selecting and utilizing these KBs. This conventional method, while effective, places limits on the flexibility and adaptability, and performance of the applications. But consider an alternative: applications that empower the LLM itself to determine which actions are necessary to generate a response. This approach not only harnesses the LLM's generative capabilities but also its ability to make contextual decisions on-the-fly, thereby opening up a more dynamic and responsive way to handle knowledge search tasks. As a result, you can build targeted KB's and use the power of LLMs to write application-specific queries that retrieve better, more relevant results. To demonstrate this approach, we'll build backed by a custom tool.
The chatbot will have access to a KB containing product information - any time you ask a question the chatbot determines could be answered using this tool it will query the KB and show both the in-progress query and retrieved results as custom UI elements before responding with the retrieved products as context. a simple chatbot We'll use Vercel's to render the chat window and Dewy to implement the information storage and retrieval used by our tool and as the knowledge store backing the tool. Generative UI library Dewy Why Dewy and Vercel? is designed to simplify the creation of custom LLM "tools" and to render rich, interactive UI elements that are specific to each tool the LLM chooses to use. Vercel's Generative UI library) What this means in practice is that your UI can dynamically adapt to the specific tools chosen by the LLM, displaying tool-specific information and progress indicators to the user. is an OSS knowledge base with the simplicity and ease-of-use of a document store: insert your documents and Dewy takes care of preparing them for semantic search.
Dewy's flexibility and ease of use allow developers to focus on building amazing user experiences rather than complex data-processing pipelines. Dewy Prerequisites Before diving into the tutorial, ensure you have the following prerequisites covered: Basic knowledge of Typescript and React A NextJS environment setup on your local machine A copy of Dewy running on your local machine (see Dewy's if you need help here). installation instructions Access to the platform. OpenAI API Set up your project : Create a NextJS app by running the following command in your terminal: Initialize a new NextJS project pnpm dlx create-next-app@canary rag-tool Navigate into your new project directory: cd rag-tool : Next, we'll install , Vercel's AI library, , OpenAI's official JavaScript SDK compatible with the Vercel Edge Runtime, and , the Dewy client library. Zod will be used to describe the input types for our custom tool. Install required packages ai openai dewy-ts pnpm install ai openai zod dewy-ts : Configure your OpenAI key and Dewy endpoint. Prepare your environment touch .env.local OPENAI_API_KEY=xxxxxxxxx
DEWY_ENDPOINT=localhost:8000 Create a custom tool First, create the custom tool the LLM will use to answer product-related questions.
This tool describes its purpose and parameters so the LLM knows when and how to use it, and defines the implementation and UI logic to complete the tool invocation. In this case, the tool will "get information about products", given a search query and a result count. When the LLM determines this tool should be used, it will generate the query string and choose how many results are appropriate, then execute the function defined below. render On execution, the tool searches Dewy for relevant products, then calls the LLM again to pick up where it left off. Since this process can take awhile, the function 's components indicating that the tool is being used and its progress. These components allow the UI to reflect the specific information relevant to this tool's outcomes. render yield import { z } from 'zod';
import { Dewy } from 'dewy-ts';
import { OpenAIStream } from 'ai'
import { Tokens } from 'ai/react';

import AssistantMessage from './AssistantMessage';
import ResultCard from './ResultCard';
import SearchCard from './SearchCard';

// Create a Dewy client.
const dewy = new Dewy({
    BASE: process.env.DEWY_ENDPOINT
})

// Implement the tool's logic.
// In this case, we search for the given query
// and return the `count` most similar chunks in the KB.
// The returned chunks are used in the `render` method below.
async function searchProducts(query: string, count: number) {
    const context = await dewy.kb.retrieveChunks({
        collection: "product_info",
        query: query,
        n: count,
    })
    return context.text_results.map(c => c.text)
}

// Define the behavior of the product search tool.
export default function productSearch(aiState, openai) {
    return {
        // A description of the tool.
        // This used by the LLM to decide when to use the tool
        description: 'Get information about products',

        // Parameters control how the tool behaves.
        // These values will be picked by the LLM,
        // so be sure to clearly explain what they're
        // used for.
        parameters: z.object({
            query: z.string().describe(```
                A description of a product 
                or what the product can be used for.
            ```),
            count: z.number().describe(```
                The number of products to return.
            ```),
        }).required(),

        // Configure the tool's behavior.
        // This function will be called after the LLM has
        // chosen to use the tool and generated values for
        // the parameters we configured above.
        render: async function* ({ query, count }) {
            // Let the user know we're looking for 
            // products related to their message
            yield <SearchCard query={query} count={count} />

            // Search for products related to the user's question
            const products = await searchProducts(query, count)

            // Update the message history 
            // with the results we found
            aiState.update([
                ...aiState.get(),
                {
                    role: "function",
                    name: "product_search",
                    content: JSON.stringify(products),
                }
            ]);

            // Now reply to the user.
            // The products we retrieved are part of the state 
            // provided as the messages parameter
            const resp = await openai.chat.completions.create({
                model: 'gpt-4-0125-preview',
                messages: aiState.get(),
                stream: true,
            })

            // Stream the results back as they're generated
            const stream = OpenAIStream(resp, {
                onFinal: (completion) => {
                    // Update the conversation history 
                    // once the full response is received
                    aiState.done([
                        ...aiState.get(),
                        {
                            role: "assistant",
                            content: completion,
                        }
                    ])
                }
            });

            // Display the response alongside 
            // the products provided to the model.
            return <div className="flex flex-col gap-2">
                <ResultCard query={query} results={context} />
                <AssistantMessage>
                    <Tokens stream={stream} />
                </AssistantMessage>
            </div>
        }
    }
} Setup the server-side message handler Configure a message handling action to receive new messages from the client.
This handler configures the initial state of the chatbot and defines the server's behavior when a message is received from the user. import { OpenAI } from 'openai';
import { createAI, createStreamableUI, getMutableAIState, render } from 'ai/rsc';

import productSearch from './productSearch';
import AssistantMessage from './AssistantMessage';

// Configure the LLM, in this case OpenAI
const openai = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY,
});
 
async function submitUserMessage(userInput: string) {
  'use server';
 
  // The AI state. This contains our message history,
  // and will be provided to the LLM each time a 
  // chat completion is generated.
  const aiState = getMutableAIState<typeof AI>();
  aiState.update([
    ...aiState.get(),
    {
      role: 'user',
      content: userInput,
    },
  ]);
 
  // Helper function for building streamable UI components.
  const ui = render({
    model: 'gpt-4-0125-preview',
    provider: openai,
    messages: [
      { role: 'system', content: `
        You are a product recommendation engine. 
        Respond only with information about products 
        retrieved using the "product_search" function role.
        `},
      { role: 'user', content: userInput }
    ],

    // This determines how generated text 
    // (as opposed to function calls) will
    // be rendered.
    text: ({ content, done }) => {
      if (done) {
        aiState.done([
          ...aiState.get(),
          {
            role: "assistant",
            content
          }
        ]);
      }
      return <AssistantMessage>{content}</AssistantMessage>
    },

    // Configure the tools available to the LLM
    tools: {
      product_search: productSearch(aiState, openai),
    }
  })
 
  return {
    id: Date.now(),
    display: ui
  };
}
 
// Define the initial state of the AI. It can be any JSON object.
const initialAIState: {
  role: 'user' | 'assistant' | 'system' | 'function';
  content: string;
  id?: string;
  name?: string;
}[] = [];
 
// The initial UI state that the client will keep track of, which contains the message IDs and their UI nodes.
const initialUIState: {
  id: number;
  display: React.ReactNode;
}[] = [{id: 0, display: <AssistantMessage>Hi! How can I help you today?</AssistantMessage>}];
 
// AI is a provider you wrap your application with so you can access AI and UI state in your components.
export const AI = createAI({
  actions: {
    submitUserMessage
  },
  // Each state can be any shape of object, but for chat applications
  // it makes sense to have an array of messages. Or you may prefer something like { id: number, messages: Message[] }
  initialUIState,
  initialAIState
}); Setup the chat UI on the client Finally, set up the chat UI on the client. Map over the UI state's messages and insert their property. display Configure a form to update the AI state, then call the server side handler. submitUserMessage 'use client'
 
import { useState } from 'react';
import { useUIState, useActions } from 'ai/rsc';
import type { AI } from './action';
import { Input } from '@/components/ui/input';
import { Button } from '@/components/ui/button';
 
import UserMessage from './UserMessage';
import { PlaneIcon } from './icons';

export default function Page() {
  const [inputValue, setInputValue] = useState('');
  const [messages, setMessages] = useUIState<typeof AI>();
  const { submitUserMessage } = useActions<typeof AI>();
 
  return (
    <>
      <div className="flex-1 overflow-auto px-4 mt-4">
        <div className="grid gap-4 md:gap-8">
          {
            messages.map((message) => (
              <div key={message.id}>
                {message.display}
              </div>
            ))
          }

        </div>
      </div>

      <form onSubmit={async (e) => {
        e.preventDefault();

        // Add user message to UI state
        setMessages((currentMessages) => [
          ...currentMessages,
          {
            id: Date.now(),
            display: <UserMessage>{inputValue}</UserMessage>,
          },
        ]);

        // Submit and get response message
        const responseMessage = await submitUserMessage(inputValue);
        setMessages((currentMessages) => [
          ...currentMessages,
          responseMessage,
        ]);

        setInputValue('');
      }}>
        <div className="border-t-2">
          <div className="flex items-center h-14 px-4">
            <Input
              className="rounded-full flex-1 min-w-0 bg-gray-200 dark:bg-gray-800"
              placeholder="Type a message..."
              value={inputValue}
              type="text"
              onChange={(event) => {
                setInputValue(event.target.value)
              }}
            />
            <Button className="ml-2 h-8 w-8 rounded-full bg-gray-200 dark:bg-gray-800" size="icon">
              <PlaneIcon className="h-4 w-4" />
              <span className="sr-only">Send message</span>
            </Button>
          </div>
        </div>
      </form>
    </>
  )
} To keep this post from going any longer than it is already, several basic UI components have been omitted, but you can checkout the full implementation in the of the Dewy repo. examples directory Conclusion In conclusion, this blog post has guided you through the process of building a Retrieval-Augmented Generation (RAG) tool using Vercel's Generative UI components. This approach moves away from the traditional methods of pre-scripting logic for selecting and utilizing knowledge bases. By empowering the LLM to determine which knowledge bases are necessary and how best to use them, we've showcased a more dynamic, efficient, and user-centric way of handling information retrieval tasks. By utilizing Dewy as our knowledge base, we've emphasized the importance of a flexible, easy-to-use platform for storing, organizing, and retrieving information. This synergy between Dewy's streamlined data management and Vercel's dynamic UI rendering paves the way for developers to create more intelligent, responsive, and user-friendly applications. Also appears . here

How to Build a "RAG Tool" With Vercel's Generative UI Components

When to Use the Logger Class? : Detailed and customizable logs simplify identifying and rectifying issues in complex applications. Debugging and Troubleshooting : Logging is essential for monitoring behavior, identifying performance bottlenecks, and analyzing usage patterns in production environments. Monitoring and Analysis : Comprehensive logging is vital for auditing and security analysis in applications handling sensitive information or needing to adhere to specific compliance requirements. Security and Compliance How to Use the Logger Class? Save the following code in a file named , and then open another file in the same root directory and use to import the package. Logger.py Python main.py from Logger import Logger Logger.py import logging
from logging import handlers

class Logger(object):
    level_relations = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'crit': logging.CRITICAL
    }  # relationship mapping

    def __init__(self, filename, level='info', when='D', backCount=3,
                 fmt='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s'):
        self.logger = logging.getLogger(filename)
        format_str = logging.Formatter(fmt)  # Setting the log format
        self.logger.setLevel(self.level_relations.get(level))  # Setting the log level
        console_handler = logging.StreamHandler()  # on-screen output
        console_handler .setFormatter(format_str)  # Setting the format
        th = handlers.TimedRotatingFileHandler(filename=filename, when=when, backupCount=backCount,encoding='utf-8')  # automatically generates the file at specified intervals
        th.setFormatter(format_str)  # Setting the format
        self.logger.addHandler(sh)  # Add the object to the logger
        self.logger.addHandler(th) This code defines a custom Logger class that can be used to create and configure loggers in Python. The class is initialized with several parameters: Logger : The name of the log file where the logs will be saved. filename : The logging level, which determines the severity of the messages to be logged (e.g., 'debug', 'info', 'warning', 'error', 'crit'). level : The interval at which the log files should be rotated (e.g., 'S' for seconds, 'M' for minutes, 'H' for hours, 'D' for days, 'W' for weeks). when : The number of backup log files to retain. backCount : The format of the log messages, which includes the time, file path, line number, log level, and the log message itself. fmt Inside the method, the logger is configured as follows: __init__ The logging level is set based on the provided . level A format string is created for formatting the log messages. A console handler is created for logging messages to the console. A rotating file handler is created for logging messages to the specified log file. Finally, the logger is set up to use both the console handler and the rotating file handler for logging messages. main.py from Logger import Logger

log = Logger('log\\all.log', level='debug')
log_error = Logger('log\\all.log', level='debug')

log.logger.info("This is an info message)")
log_error.info("This is an error message") This Python code shows how to use the Logger package to write log information to a file. It first initializes two Logger objects, one for general logging and one for error logging. It then uses these objects to write log messages at different levels, such as info and error. The log messages are written to the specified log file, allowing for easy monitoring and troubleshooting of the application. Customizing Parameters The Logger class provides several parameters that can be customized based on the specific logging needs. These parameters include: : The name of the log file where the logs will be written. filename : The log level, which determines the severity of the logs (e.g., debug, info, warning, error, crit). level : The frequency at which the log files should be rotated (e.g., 'D' for daily, 'H' for hourly). when : The number of backup log files to retain after rotation. backCount : The format of the log messages, allowing customization of the timestamp, log level, file name, line number, and message content. fmt By adjusting these parameters, the Logger class can be tailored to meet the specific logging requirements of different projects and applications. Example 1: Customizing Log Level and Format log = Logger('custom.log', level='warning', fmt='%(asctime)s - %(levelname)s - %(message)s')
log.logger.debug("This is a debug message - won't be logged")
log.logger.warning("This is a warning message - will be logged") Example 2: Specifying Log Rotation Parameters log = Logger('rotation.log', when='midnight', backCount=7)
log.logger.info("Logging with rotation based on time") Conclusion Logging is crucial in software development for monitoring, troubleshooting, analyzing, and securing applications. Python's logging class offers a convenient and customizable way to implement logging with various configuration options. This class enables developers to tailor logging setups to meet project needs, ensuring comprehensive and efficient logging for debugging, monitoring, and compliance purposes. The Logger class empowers developers to harness the power of logging for robust application development and maintenance. Happy logging! Thank you for taking the time to explore data-related insights with me. I appreciate your engagement. If you find this information helpful, I invite you to follow me or connect with me on . Happy exploring!👋 LinkedIn Also published . here

Python Logging: Debugging, Monitoring, and Security with the Logger Class

Introduction You’ve heard of data warehouses, you’ve probably heard of data lakes and the data lakehouse, but have you heard of the “Streamhouse”? Well, that’s what “they” call the new data tech unlock possible with . So, just what the heck is it? Let’s find out! Apache Paimon A company named Ververica, the original creator of , looks pretty excited about Paimon, but Jingsong Lee primarily created it at Alibaba. The purpose initially was to address the inability to query a Flink Dynamic Table, which resulted in the community proposing . Apache Flink FLIP-188: Introduce Built-in Dynamic Table Storage Paimon can be thought of as a table format for batch and stream processing. I’ve written a couple of times about the big three table formats and . here here Paimon Overview The short version of what Paimon does is that it provides a streaming storage layer and extends Flink's capability to do stream processing directly on the data lake. The use case seems to be tightly coupled to Flink at the moment by giving you a more flexible and persistent storage layer for your streaming data. Instead of using the ‘InMemory Catalog’, you would set up a ‘Paimon Catalog’ like other table formats. That would look something like this: CREATE CATALOG paimon WITH (
   'type' = 'paimon',
   'Warehouse' = '<path to your warehouse>'
);
USE CATALOG PAIMON; To explain the flow, I’m going to borrow a screenshot from the by Giannis Polyzos, excellent blog If you aren’t familiar with what an LSM Tree (Log-Structured Merge) is, then I suggest by Vishal Rana. I think he did a nice job explaining it. this blog So, Paimon can be used to replace message queues, providing a storage layer for Flink that is a table format. As a streaming data lake platform, it allows users to process data in both batch and streaming modes, supporting: High-speed data ingestion Change data tracking Real-time analytics High throughput data writing Low-latency data queries Batch writes and reads Streaming updates Changelog producing Where It Applies The first few use cases that occurred to me were gaming, IoT devices, stock trading, and services like Uber. There is generally a lot of data coming in quickly, and you probably need to be able to do something actionable from it. I remember having to deal with combining streaming data from ad impressions and ad clicks to get insights on advertising data as another example. Having this kind of a materialized view and a table format defined on it would certainly have been helpful. I bet you can think of some use cases that would help you out. Borrowing another screenshot from Giannis Polyzos, we see how we are getting to what is being called a “Streamhouse.” I found this very interesting after my time in the (compute engine) and (table format) communities. PrestoDB Apache Iceberg Streaming would come up fairly often with Iceberg, and there is an argument out there that the Apache Hudi table format is better for that use case. Paimon can also have some issues with Change Data Capture (CDC) ingestion into a data lake. It simplifies the CDC pipeline, supporting synchronizing CDC data with schema changes, streaming changelog tracking, and a partial-update merge engine. Summary As I researched for this blog, I thought Paimon and Seatunnel would make an exciting combination for a real-time data warehouse. Then I ran across this blog, “ ” from back in May 2023, so clearly, I’m not the first to think about this idea. Apache SeaTunnel and Paimon: Unleashing the Potential of Real-Time Data Warehousing With these new projects, we’re seeing an interesting shift in the streaming and real-time space, and I’m here for it. I wish I personally had a use case to try these things out on a big, live system. We’re seeing more and more clever technology being developed to address new problems that arise. I wonder how the Amazon S3 Express One Zone Storage Class announcement might impact the “real-time” space. It’s all very exciting to watch. You can read more "What the heck" articles at the following links: What The Heck Is DuckDB? What the Heck Is Malloy? What the Heck is PRQL? What the Heck is GlareDB? What the Heck is SeaTunnel? What the Heck is LanceDB? What the Heck is SDF?

What the Heck Is Apache Paimon?

Kafka is widely used for building event-driven architectures, allowing services to communicate through messages. However, when working with a foreign provider that adds a " " header to messages, you may encounter serialization and deserialization issues. microservices TypeId This article will guide you through the process of configuring a custom type mapper in Spring Kafka to handle the mapping of types between the producer and consumer services. By implementing this approach, you can successfully consume messages with different package structures and types, avoiding exceptions caused by incompatible types and ensuring smooth message processing. Understanding the deserialization Exception with " " Header TypeId In scenarios where the classes used in the consumer service have a different package structure compared to the producer service's DTO classes, the serialization exception can occur due to the mismatched class paths. DTO The serialization process relies on both the producer and the consumer having access to the same class definitions. When the consumer receives a message with a " " header referencing a class that resides in a different package than the consumer's DTO classes, the deserialization process fails to locate the appropriate class definition. This results in the "ClassNotFoundException" or "IllegalArgumentException" within the serialization exception. TypeId These exceptions can disrupt the smooth consumption of messages. Understanding the root cause of this exception is key to effectively handling the situation. An example of the serialization exception is as follows: javaCopy codejava.lang.IllegalStateException: This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer
	at org.springframework.kafka.listener.DefaultErrorHandler.handleOtherException(DefaultErrorHandler.java:198) ~[spring-kafka-3.0.7.jar:3.0.7]
Caused by: org.apache.kafka.common.errors.RecordDeserializationException: Error deserializing key/value for partition consumer-test-18 at offset 1. If needed, please seek past the record to continue consumption.
	at org.apache.kafka.clients.consumer.internals.Fetcher.parseRecord(Fetcher.java:1438) ~[kafka-clients-3.4.0.jar:na]
Caused by: java.lang.IllegalArgumentException: The class 'org.example.dto.out.OutMessageImplTwo' is not in the trusted packages: [java.util, java.lang, org.example.dto.in, org.example.dto.in.*]. 
If you believe this class is safe to deserialize, please provide its name. If the serialization is only done by a trusted source, you can also enable trust all (*).
	at org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:129) ~[spring-kafka-3.0.7.jar:3.0.7] The exception message provides insights into the underlying problem. In this case, the exception occurs because the class is not present in the list of trusted packages. The trusted packages serve as a safeguard to ensure secure deserialization and prevent unauthorized classes from being deserialized. org.example.dto.out.OutMessageImplTwo When deserializing a message, the default behavior of Spring Kafka is to check if the class specified in the "TypeId" header is within the trusted packages. If it is not found, an exception is thrown. This mechanism is in place to protect against potential security risks, as deserializing untrusted classes can lead to code execution vulnerabilities. To resolve this issue, you need to ensure that the consumer can properly map the incoming message to the corresponding class, even if they have different package structures. One approach is to configure a custom type mapper that explicitly handles the type mapping between the " " header and the consumer's DTO classes. TypeId By implementing a custom type mapper, you can instruct the consumer to map the " " header to the appropriate class or even ignore this header. This enables successful deserialization and ensures that the consumer can process the message correctly. TypeId In the upcoming sections, we will delve into the details of configuring and implementing a custom type mapper in Kafka listeners, allowing seamless message processing and addressing the challenges encountered when the consumer's DTO classes differ from those of the producer. Implementation of custom-type mapper In our example, we will send an implementation of OutMessageAbstract to Kafka and will read it as an implementation of the InMessageAbstract class. These classes are absolutely identical, they differ in name and package. The figure below shows the project structure. Our KafkaListener looks like this. It implements the logic for processing each type of message from the topic. @Component
@Log
@KafkaListener(
        topics = "${spring.cloud.stream.bindings.topic-in-0.destination}",
        groupId = "${spring.cloud.stream.bindings.topic-in-0.group}",
        containerFactory = "kafkaListenerContainerFactory")
public class EventListeners {
    @KafkaHandler
    public void handleInMessageImplEvent(@Payload InMessageImpl event) {
        System.out.println(event);
    }

    @KafkaHandler
    public void handleInMessageImplTwoEvent(@Payload InMessageImplTwo event) {
        System.out.println(event);
    }
} We use the kafkaListenerContainerFactory bean to configure our listener. This factory is described in KafkaConsumerConfiguration and looks like this: @Configuration
public class KafkaConsumerConfiguration {
    private final String SERVER;
    private final String SERVER_PORT;

    public KafkaConsumerConfiguration(
            @Value("${spring.cloud.stream.kafka.binder.brokers}") String server,
            @Value("${spring.cloud.stream.kafka.binder.defaultBrokerPort}") String port
    ) {
        this.SERVER = server;
        this.SERVER_PORT = port;
    }

    private <T> ConsumerFactory<String, T> typeConsumerFactory(Class<T> clazz) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, String.format("%s:%s", SERVER, SERVER_PORT));

        return new DefaultKafkaConsumerFactory<>(
                props,
                new StringDeserializer(),
                new JsonDeserializer<>(clazz));
    }

    private <T> ConcurrentKafkaListenerContainerFactory<String, T> initFactory(Class<T> clazz) {
        ConcurrentKafkaListenerContainerFactory<String, T> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(typeConsumerFactory(clazz));
        return factory;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, ?> kafkaListenerContainerFactory() {
        return initFactory(InMessageAbstract.class);
    }
} Because we're adding a “Typeld” header to the message, we'll get the error described above when we try to deserialize. However, we can configure the JsonDeserializer manually. To do this, we must define our implementation of TypeMapper. The following shows how you can define your own TypeMappers. class CustomTypeMapper<T> extends DefaultJackson2JavaTypeMapper {
    private final Class<T> clazz;

    public CustomTypeMapper(Class<T> clazz) {
        this.clazz = clazz;
    }

    @Override
    public JavaType toJavaType(Headers headers) {
        try {
            return TypeFactory.defaultInstance()
                    .constructType(ClassUtils.forName(clazz.getName(), getClassLoader()));
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
} If we add our TypeMapper implementation to the JsonDeserializer when we create the factory, then we can force the Json to be deserialized to the class we need, ignoring the header. The corrected method would look like this: private <T> ConsumerFactory<String, T> typeConsumerFactory(Class<T> clazz) {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, String.format("%s:%s", SERVER, SERVER_PORT));

    JsonDeserializer<T> jsonDeserializer = new JsonDeserializer<>(clazz);
    jsonDeserializer.setTypeMapper(new CustomTypeMapper<>(clazz));

    return new DefaultKafkaConsumerFactory<>(
            props,
            new StringDeserializer(),
            jsonDeserializer);
} Conclusion By utilizing a custom type mapper in Spring Kafka, you can overcome the challenges associated with consuming messages from foreign providers that include a " " header. This article has provided a step-by-step guide to configuring and implementing the custom type mapper, enabling smooth message processing and avoiding serialization and deserialization exceptions. With this knowledge, you can confidently integrate with external systems and consume Kafka messages with different package structures. TypeId The full project code is available on GitHub: https://github.com/stepanovD/kafka-templates/tree/master/consumer-custom-type-mapper

Handling Custom Type Mapping in Kafka Listeners for Messages with "TypeId" Header

The library to be introduced - - is a Kafka Streams State Store implementation that persists data to . Java thriving-dev/kafka-streams-cassandra-state-store Apache Cassandra It's a 'drop-in' replacement for the official Kafka Streams state store solutions, notably (default) and . RocksDB InMemory By moving the state to an datastore the (from a deployment point of view) - which greatly . external stateful streams app effectively becomes stateless improves elasticity, and reduces rebalancing downtimes & failure recovery Cassandra/ScyllaDB is horizontally scalable and allows for which provides a boost to your existing Kafka Streams application with very little change to the source code. huge amounts of data In addition to the this post will also cover all out-of-the-box state store solutions, and explain individual characteristics, benefits, drawbacks, and limitations in detail. CassandraKeyValueStore ⏯️ Following the introduction and getting started guide, there's also a available. If you don't want to wait, feel free to head over to the . demo Thriving.dev YouTube Channel The first public release was on 9 January 2023. When writing this blog post the latest version was: - available on ! 0.4.0 Maven Central Table of contents Basics Recap Purpose Usage Example Under the Hood Known Limitations Next Steps Conclusion Basics Recap (Feel free to skip straight to the next section if you're already familiar with Kafka Streams and Apache Cassandra…) Kafka Streams Quoting : Apache Kafka - Wikipedia “Kafka Streams (or Streams API) is a stream-processing library written in Java. It was added in the Kafka 0.10.0.0 release. The library allows for the development of stateful stream-processing applications that are scalable, elastic, and fully fault-tolerant. The main API is a stream-processing  domain-specific language  (DSL) that offers high-level operators like filter,  map, grouping, windowing, aggregation, joins, and the notion of tables. Additionally, the Processor API can be used to implement custom operators for a more low-level development approach. The DSL and Processor API can be mixed, too. For stateful stream processing, Kafka Streams uses  RocksDB  to maintain local operator state. Because RocksDB can write to disk, the maintained state can be larger than available main memory. For fault-tolerance, all updates to local state stores are also written into a topic in the Kafka cluster. This allows recreating state by reading those topics and feed all data into RocksDB.” In case you are entirely new to Kafka Streams, I recommend getting started by reading some official materials provided by Confluent, e.g. Introduction Kafka Streams API Apache Cassandra Quoting : Apache Cassandra - Wikipedia “Apache Cassandra is a free and open-source, distributed, wide-column store, NoSQL database management system designed to handle large amounts of data across many commodity servers, providing high availability with no single point of failure. Cassandra offers support for clusters spanning multiple datacenters, with asynchronous masterless replication allowing low latency operations for all clients.” Purpose While Wikipedia’s summary (see above) only mentions RocksDB, Kafka Streams ships with following implementations: KeyValueStore org.apache.kafka.streams.state.internals.RocksDBStore org.apache.kafka.streams.state.internals.InMemoryKeyValueStore org.apache.kafka.streams.state.internals.MemoryLRUCache Let’s look at the traits of each store implementation in more detail… RocksDBStore RocksDB is the default state store for Kafka Streams. The is a key-value store based on (surprise!). State is flushed to disk, allowing the state to exceed the size of available memory. RocksDBStore persistent RocksDB Since the state is persisted to disk, it can be re-used and does not need to be restored (changelog topic replay) when the application instance comes up after a restart (e.g. following an upgrade, instance migration, or failure). The RocksDB state store provides good performance and is well configured out of the box, but might need to be tuned for certain use cases (which is no small feat and requires an understanding of RocksDB configuration). Writing to and reading from disk comes with I/O, for performance reasons buffering and caching patterns are in place. The record cache (on heap) is particularly useful for optimizing writes by reducing the number of updates to local state and changelog topics. The RocksDB block cache (off heap) optimizes reads. In a typical modern setup, stateful Kafka Streams applications run on Kubernetes as a with persistent state stores (RocksDB) on . StatefulSet PersistentVolumes InMemoryKeyValueStore The , as the name suggests, maintains state (RAM). InMemoryKeyValueStore in-memory One obvious benefit is that the pure in-memory stores come with good performance (operates in RAM…). Further, hosting and operating are simpler compared to RocksDB, since there is no requirement to provide and manage disks. Drawbacks to having the store in-memory are limitations in store size and increased infrastructure costs (RAM is more expensive than disk storage). Further, state always is lost on application restart and therefore first needs to be restored from changelog topics (recovery takes longer). When low rebalance downtimes / quick recovery is concerned, using standby replicas ( ) help to reduce recovery time. num.standby.replicas MemoryLRUCache ( ) Stores.lruMap The is an store based on . The term comes from the (least recently used) behavior combined with the cap (per streams task!). MemoryLRUCache in-memory HashMap cache LRU maxCacheSize It’s a rather uncommon choice but can be a valid fit for certain use cases. Same as the state always is lost on application restart and is restored from changelog topics. InMemoryKeyValueStore applies client-side only (in-memory HashMap, per streams task state store -> the least recently used entry is dropped when the underlying HashMap’s capacity is breached) but does not ‘cleanup’ the changelog topic (send ). The ( ) changelog topic keeps growing in size while the state available to processing is constrained by . maxCacheSize tombstones compacted maxCacheSize Therefore, it is recommended to use in combination with custom changelog topic config (also ) to have a time-based retention in place that satisfies your functional data requirements (if possible). cleanup.policy=[compact,delete] retention.ms The is applied per streams task (~input topic partitions), so take into consideration when calculating total capacity, memory requirements per app instance, … maxCacheSize CassandraKeyValueStore Now finally we get to the subject of this blog post, the custom implementation of a state store that persists data to Apache Cassandra. With data is in an external database -> Apache Cassandra <- *or compatible solutions (e.g. ScyllaDB). Apache Cassandra is a distributed, clustered data store that allows to scale horizontally to enable up to Petabytes of data, thus can be accommodated. CassandraKeyValueStore persistently stored very large Kafka Streams state Moving the state into an data store - outside the application so to say - allows you to effectively run the app in a fashion. Further, with logging disabled, there's -> required which enables fluent rebalancing and helps and . external stateless no changelog topic no state restore reduce rebalance downtimes recovery time This greatly of your application, which opens up more possibilities such as e.g. efficient & fluent autoscaling... improves the elasticity and scalability It can also help ease/avoid known problems with the 'Kafka Streams’ specific task assignment such as 'uneven load distribution' and 'idle consumers' (I'm thinking about writing a separate blog post on these issues...). Kafka Streams property allows to achieve low rebalance downtimes by telling the consumers to send a request to the group leader on graceful shutdown. internal.leave.group.on.close=true LeaveGroup For more information on such kafka internals I can recommend to watch+read following Confluent developer guide: . Consumer Group Protocol Note that this property is also used & explained in the . demo ⚠ Please be aware this is an in-official property (not part of the public API), thus can be deprecated or dropped any time. Adding an external, 3rd party Software to the heart (or rather stomach?) to your stream processing application, adds a new, additional to your architecture. single point of failure Usage Example Get it! The artifact is available on Maven Central: Maven <dependency>
    <groupId>dev.thriving.oss</groupId>
    <artifactId>kafka-streams-cassandra-state-store</artifactId>
    <version>${version}</version>
</dependency> Gradle (Groovy DSL) implementation 'dev.thriving.oss:kafka-streams-cassandra-state-store:${version}’ Classes of this library are in the package . dev.thriving.oss.kafka.streams.cassandra.state.store Quick Start High-level DSL <> StoreSupplier When using the high-level DSL, i.e., , users create s that can be further customized via . StreamsBuilder StoreSupplier Materialized For example, a topic read as can be materialized into a Cassandra k/v store with custom key/value Serdes, with logging and caching disabled: KTable StreamsBuilder builder = new StreamsBuilder();
KTable<Long,String> table = builder.table(
  "topicName",
  Materialized.<Long,String>as(
                 CassandraStores.builder(session, "store-name")
                         .keyValueStore()
              )
              .withKeySerde(Serdes.Long())
              .withValueSerde(Serdes.String())
              .withLoggingDisabled()
              .withCachingDisabled()); Processor API <> StoreBuilder When using the Processor API, i.e., , users create s that can be attached to s. Topology StoreBuilder Processor For example, you can create a Cassandra with custom Serdes, logging, and caching disabled like: KeyValueStore<String, Long> Topology topology = new Topology();

StoreBuilder<KeyValueStore<String, Long>> storeBuilder = Stores.keyValueStoreBuilder(
                CassandraStores.builder(session, "store-name")
                        .keyValueStore(),
                Serdes.String(),
                Serdes.Long())
        .withLoggingDisabled()
        .withCachingDisabled();

topology.addStateStore(storeBuilder); Demo Features the notorious ( ), written as a , running in a fully . 'word-count example' ref quarkus application clustered docker-compose localstack Source code for this demo: kafka-streams-cassandra-state-store/examples/word-count-quarkus (at 0.4.0) https://www.youtube.com/watch?v=2Co9-8E-uJE&embedable=true Store Types kafka-streams-cassandra-state-store comes with 2 different store types: keyValueStore globalKeyValueStore keyValueStore (recommended default) A persistent . The underlying cassandra table is the store context . Therefore, all CRUD operations against this store always query by and return results for a single stream task. KeyValueStore<Bytes, byte[]> partitioned by task partition globalKeyValueStore A persistent . The underlying cassandra table uses the . Therefore, all CRUD operations against this store work from any streams task and therefore always are “global”. KeyValueStore<Bytes, byte[]> record key as sole /PRIMARY KEY/ Due to the nature of cassandra tables having a single PK (no clustering key), this store supports only a limited number of operations. If you're planning to use this store type, please make sure to get a full understanding of the specifics by reading the to understand its behaviour. relevant docs Advanced For more detailed documentation, please visit the GitHub project… Store types, supported operations Builder usage + config options Under the hood Implemented/compiled with Java 17 kafka-streams 3.4 datastax java-driver-core 4.15.0 Supported client-libs Kafka Streams 2.7.0+ (maybe even earlier versions, but wasn’t tested further back) Datastax java client (v4) 'com.datastax.oss:java-driver-core:4.15.0' ScyllaDB shard-aware datastax java client (v4) fork 'com.scylladb:java-driver-core:4.14.1.0' Supported databases Apache Cassandra 3.11 Apache Cassandra 4.0, 4.1 ScyllaDB (should work from 4.3+) Underlying CQL Schema keyValueStore Using defaults, for a state store named "word-count" the following CQL Schema applies: CREATE TABLE IF NOT EXISTS word_count_kstreams_store (
    partition int,
    key blob,
    time timestamp,
    value blob,
    PRIMARY KEY ((partition), key)
) WITH compaction = { 'class' : 'LeveledCompactionStrategy' } globalKeyValueStore Using defaults, for a state store named "clicks-global" the following CQL Schema applies: CREATE TABLE IF NOT EXISTS clicks_global_kstreams_store (
    key blob,
    time timestamp,
    value blob,
    PRIMARY KEY (key)
) WITH compaction = { 'class' : 'LeveledCompactionStrategy' } Feat: Cassandra table with default TTL Cassandra has a table option (default expiration time (“TTL”) in seconds for a table) which can be useful for certain use cases where data (state) expires after a known period. default_time_to_live ℹ️ that writes to Cassandra are made with system time. The table based on -> the time of the current record being processed ( ). Please note TTL is applied 'time of write' != stream time The can be defined via the builder method, e.g.: default_time_to_live withTableOptions CassandraStores.builder(session, "word-grouped-count")
        .withTableOptions("""
                compaction = { 'class' : 'LeveledCompactionStrategy' }
                AND default_time_to_live = 86400
                """)
        .keyValueStore() Cassandra table partitioning (avoiding large partitions) Kafka is persisting data in segments and is built for sequential r/w. As long as there’s sufficient disk storage space available to brokers, a high number of messages for a single topic partition is not a problem. Apache Cassandra on the other hand can get inefficient (up to severe failures such as load shedding, dropped messages, and crashed and downed nodes) when the partition size grows too large. The reason is that searching becomes too slow as the search within a partition is slow. Also, it puts a lot of pressure on the (JVM) heap. The community has offered a standard recommendation for Cassandra users to keep Partitions under 400MB, and preferably under 100MB. For the current implementation, the Cassandra table created for the ‘default’ key-value store is partitioned by the Kafka (“wide partition pattern”). partition key Please keep these issues in mind when working with relevant data volumes. In case you don’t need to query your store / only lookup by key (‘range’, ‘prefixScan’; ref ) it’s recommended to use rather than since it is partitioned by the (:= primary key). Supported operations by store type globalKeyValueStore keyValueStore event key References blog post on Wide Partitions in Apache Cassandra 3.11 in case anyone has funded knowledge if/how this has changed with Cassandra 4, please share in the comments below!! Note: stackoverflow question Known Limitations Adding additional infrastructure for data persistence external to Kafka comes with certain risks and constraints. Consistency Kafka Streams supports and processing guarantees. At-least-once semantics is enabled by default. at-least-once exactly-once Kafka Streams processing guarantees is using Kafka transactions. These transactions wrap the entirety of processing a message throughout your streams topology, including messages published to outbound topic(s), changelog topic(s), and consumer offsets topic(s). exactly-once This is possible through transactional interaction with a single distributed system (Apache Kafka). Bringing an external system (Cassandra) into play breaks this pattern. Once data is written to the database it can’t be rolled back in the event of a subsequent error / failure to complete the current message processing. => If you need strong consistency, have processing enabled (streams config: ), and/or your processing logic is not fully idempotent then using is discouraged! exactly-once processing.guarantee="exactly_once_v2" kafka-streams-cassandra-state-store ℹ️ Please note this is also the case when using kafka-streams with the native state stores (RocksDB/InMemory) with processing.guarantee (default). at-least-once For more information on Kafka Streams processing guarantees, check the sources referenced below. https://medium.com/lydtech-consulting/kafka-streams-transactions-exactly-once-messaging-82194b50900a https://docs.confluent.io/platform/current/streams/developer-guide/config-streams.html#processing-guarantee https://docs.confluent.io/platform/current/streams/concepts.html#processing-guarantees Incomplete Implementation of Interfaces StateStore For now, only is supported (vs. e.g. / ).
Also, not all methods have been implemented. KeyValueStore WindowStore SessionStore Next Steps Here are some of the tasks (high level) in the current backlog: Features Implement (coming soon with Kafka 3.5.0 release) KIP-889: Versioned State Stores Add a simple (optional) InMemory read cache -> ? Caffeine Support / WindowStore SessionStore Non-functional Benchmark Add metrics Ops GitHub actions to release + publish to maven central (snapshot / releases) Add Renovate Conclusion It's been a fun journey so far, starting from an initial POC to a working library published to maven central - though still to be considered 'experimental', since it's not been production-tested yet. The out-of-the-box state stores satisfy most requirements, no need to switch without necessity.
Still, it's a usable piece of software that may fill a gap for specific requirements. I'm looking forward to working on the next steps such as benchmarking/ . load testing Feedback is very welcome, also, if you are planning to, or have decided to use the library in a project, please leave a comment below. Footnotes At the time of writing this blog post, the latest versions of relevant libs were: Kafka / Streams API: 3.4.0 Cassandra java-driver-core: 4.15.0 kafka-streams-cassandra-state-store: 0.4.0 References 4. Stateful Processing - Mastering Kafka Streams and ksqlDB Book Kafka Streams Memory Management | Confluent Documentation Also published . here

Unveiling Kafka-Streams-Cassandra-State-Store

I've blogged quite a bit about the Amazon Interactive Video Service (Amazon IVS) Web Broadcast SDK. We've learned about , how to use the SDK to , how to , and even looked at using it to . the basics stream pre-recorded videos add screen sharing and overlays create a Lofi radio station The Web Broadcaster SDK is a game changer that gives developers the ability to integrate the broadcast experience directly into their streaming applications instead of directing their users to use third-party desktop streaming software. In this post, we'll look at another exciting possible use case of the Web Broadcast SDK: streaming browser based games directly to an Amazon IVS channel. It's really quite easy to do, and we'll take it a step further by creating an interactive experience that allows live stream viewers to control the gameplay (akin to ). TwitchPlaysPokemon Streaming a Browser Based Game Directly From the Browser For this post, I'll assume that you're already familiar with Amazon IVS and have already configured a live streaming channel. If you're new to Amazon IVS, check out the blog series , specifically the very . Optionally, you could also refer to the which is a great resource for learning how to develop live streaming applications with Amazon IVS. Getting Started with Amazon IVS first post user guide Since I'm not a game developer, I decided to add live streaming to a few existing open source, browser-based games: and . Since both of these games utilize for gameplay, it will be easy to obtain a from them will be the source for our live stream video input. pacman-canvas Astray <canvas> MediaStream pacman-canvas For the first demo, I cloned the repo to my local machine and took a look at the code. pacman-canvas Initializing the Broadcast Client The game uses jQuery, so I added a call to an method to the end of the existing DOM ready handler: pacman-canvas initBroadcast() let broadcastClient;
let isBroadcasting = false;

$(document).ready(function () {
  // game logic...
  initBroadcast();
}) In my method, I create an instance of the ( ), passing it the from my Amazon IVS channel. initBroadcast() AmazonIVSBroadcastClient docs Ingest endpoint broadcastClient = IVSBroadcastClient.create({
  streamConfig: IVSBroadcastClient.STANDARD_LANDSCAPE,
  ingestEndpoint: config.ingestEndpoint,
}); Next, to add the gameplay to the client, I grabbed a reference to the element used by the game and called ( ) on the : <canvas> addVideoInputDevice() docs broadcastClient const game = document.getElementById('myCanvas');
broadcastClient.addVideoInputDevice(game.captureStream(), 'game-track', { index: 0 }); Finally, to start the broadcast, I call and pass it the from my channel. startBroadcast() Stream key broadcastClient
  .startBroadcast(config.streamKey)
  .then(() => {
    isBroadcasting = true;
    console.log('online')
  })
  .catch((error) => {
    isBroadcasting = false;
    console.error(error);
  }); The game uses a few tags for sound effects and calls the following function as necessary. <audio> var Sound = {};
Sound.play = function (sound) {
  if (game.soundfx == 1) {
    var audio = document.getElementById(sound);
    (audio !== null) ? audio.play() : console.log(sound + " not found");
  }
}; To add the game audio to my stream, I modified it to call ( ). addAudioInputDevice() docs var Sound = {};
Sound.play = function (sound) {
  if (game.soundfx == 1) {
    var audio = document.getElementById(sound);
    (audio !== null) ? audio.play() : console.log(sound + " not found");
    var trackLabel = `${sound}-audio-track-${new Date().getTime()}`;
    audio.addEventListener('playing', (evt) => {
      if (!broadcastClient.getAudioInputDevice(trackLabel)) {
        broadcastClient.addAudioInputDevice(audio.captureStream(), trackLabel);
      }
    });
  }
}; And that's it! The gameplay stream will be broadcast whenever the method is called, so we could attach that to a button click handler or call it from the method of the existing game. If we wanted to, we could also and even size and position the webcam as an overlay (see ( ) for more info). startBroadcast() newGame add a webcam to the stream VideoComposition docs Playing Back the Live Stream in a Browser For playback, I added a simple HTML page that includes and utilizes the . Amazon IVS Player SDK <script src="https://player.live-video.net/1.17.0/amazon-ivs-player.min.js"></script>
<script>
document.addEventListener('DOMContentLoaded', () => {
  const videoPlayer = document.getElementById('video-player');
  const streamUrl = '[CHANNEL PLAYBACK URL]';
  const ivsPlayer = IVSPlayer.create(); 
  ivsPlayer.attachHTMLVideoElement(videoPlayer);
  ivsPlayer.load(streamUrl);
  ivsPlayer.play();
});
</script>
<body>
  <video id="video-player" />
</body> Here's how everything looks at this point. On the left side is the broadcast/gamer view, and on the right is the playback/viewer view. Adding Timed Metadata As you can see in the Gif above, the score, level, and lives data is not included in the live stream. This is because the game developer did not include those elements in the gameplay . If we wanted to, we could render these as overlays in the playback view by . In my demo, I modified the function of to publish the score as timed metadata on my live stream by calling an AWS Lambda function that I've previously created for this channel. <canvas> publishing them with timed metadata Score pacman-canvas const publishMetadata = async (meta) => {
  await fetch('[lambda url]/send-metadata', {
    method: 'POST',
    headers: {
      'Accept': 'application/json',
      'Content-Type': 'application/json'
    },
    body: JSON.stringify(meta),
  });
};

function Score() {
  this.score = 0;
  this.set = function (i) {
    this.score = i;
  };
  this.add = function (i) {
    this.score += i;
    publishMetadata({
      metadata: JSON.stringify({ score: this.score }),
    });
  };
  this.refresh = function (h) {
    $(h).html("Score: " + this.score);
  };

} Be careful publishing metadata too frequently to avoid exceeding the Amazon IVS . You may need to throttle or batch metadata publishing to avoid exceeding the quota. Note: service quotas Adding Interactivity to a Browser Based Game Live Stream Now let's take a look at how we might create an interactive browser based game that can be controlled by live stream viewers. For this, I chose - a simple game that requires the player to navigate a ball through a maze. I added the Web Broadcast client just as i did with above, and decided to use Amazon IVS chat to receive the gameplay control commands from the viewer. Astray pacman-canvas Adding Chat We've previously looked at how to . To use chat for game interactivity, we'll configure a chat experience just like we usually do, but in the chat connection's message handler, we'll check the message content for the terms , , and to control the ball movement on the broadcaster side. add chat to your live streaming application with Amazon IVS chat left right up down const chatConnection = new WebSocket('wss://edge.ivschat.us-east-1.amazonaws.com', token);

chatConnection.onmessage = (event) => {
  const data = JSON.parse(event.data);
  const chatEl = document.getElementById('chat');
  if (data.Type === 'MESSAGE') {
    const direction = data.Content.toLowerCase();
    if (['left', 'up', 'right', 'down'].indexOf(direction) > -1) moveBall(direction);
    // render message to chat div
  }
} If I find one of the directional commands in the incoming message, I call which will simulate a key press on the broadcaster side to move the ball in the game. moveBall() const moveBall = (direction) => {
  let keyCode;
  let keyCodes = {
    left: 37,
    up: 38,
    right: 39,
    down: 40,
  }
  const keyDown = new KeyboardEvent('keydown', { keyCode: keyCodes[direction] });
  const keyUp = new KeyboardEvent('keyup', { keyCode: keyCodes[direction] });
  document.dispatchEvent(keyDown);
  setTimeout(() => {
    document.dispatchEvent(keyUp);
  }, 200);
}; There's a minor catch to this approach. Because uses , the broadcaster's game tab must be active/visible tab at all times since to improve performance and battery life. Note: Astray requestAnimationFrame() requestAnimationFrame() pauses when the tab is not visible Here's how looks when controlled by the viewer. Astray There's a slight delay due to the stream latency, but the potential to give viewers control of the live stream gameplay is intriguing and filled with potential. Summary In this post, we learned how to integrate the Amazon IVS Web Broadcast SDK into a browser based game to give players the ability to live stream gameplay directly to an Amazon IVS live streaming channel. We also learned how to add interactivity to the experience to give stream viewers the ability to directly affect the gameplay. If you'd like to learn more about Amazon IVS, check out the blog post series here on HackerNoon. Getting Started with Amazon IVS Also published . here

How to Broadcast Interactive Web Based Gaming Live Streams with Amazon IVS

Sometimes it's the little features that differentiate one application from another. If you've added live streaming to your application and your user's navigate to a channel that they are interested in viewing, what happens if the channel is not currently broadcasting when they hit the page? Maybe you have added to create an interactive experience so that viewers can spend the time waiting for a broadcast chatting with each other, but what happens when the broadcast goes live? Do your viewers have to click a 'play' button, or worse yet reload the entire web page to view the stream? How can we improve this experience? Wouldn't it be better to autoplay the stream when it goes live? Of course it would! Sadly, the Amazon IVS does not have a built-in method for autoplaying live stream. But, we can work around this limitation! Amazon Interactive Video Service (Amazon IVS) chat player One option to autoplay videos is to use polling (via on the client side) to check to see if the stream is broadcasting: setInterval() setInterval(() => {
  ivsPlayer.load(streamUrl);
  ivsPlayer.play();
}); This technically , but it's not the best idea. With the polling approach, you would end up with tons of errors in the console. Also, you'd have to store the interval and eventually clear it later on when the stream starts playing. How often should you run the check? Every second seems to frequent, every 10 seconds seems to infrequent. It's a messy approach, and requires extra code to manage. works 404 A better approach would be to send a push notification to the client when a stream goes online. With this approach, you avoid the messiness of polling and can be assured that the stream starts as soon as possible after the broadcaster has gone online. So how do we know when a stream is online, and how can we easily send the notification to the client? We've previously looked at using or alerts when a stream goes online, and we can use a similar approach to autoplay live streams. With this approach, we'll create an EventBridge rule that triggers an AWS Lambda function when a stream goes online. But, instead of (or in addition to) sending a push notification, we can use Amazon IVS chat custom events to send a message to all connected chat clients and in the client-side handler start playing the stream when the message is received. Amazon EventBridge rules to send push notifications That's OK! You can utilize any pub/sub or WebSocket implementation to send the message to your front end. Also, even if you're not using Amazon IVS chat for interactive chat, you can still utilize it as an event bus for application messaging. Not using Amazon IVS Chat? Creating the Lambda Function and EventBridge Rule Let's set up a serverless application using AWS Serverless Application Model (AWS SAM). In this application, we'll have a function that will be used to send the custom chat event. Let's first look at the AWS Lambda function ( ). This function will need to use a layer that enables us to utilize the module from the AWS SDK for JavaScript (v3). We'll use the ( ) method to send a custom event to our chat clients. index.mjs @aws-sdk/ivs-chat SendEvent docs import { IvschatClient, CreateChatTokenCommand, SendEventCommand } from '@aws-sdk/client-ivschat';

const chatArn = process.env.DEMO_CHAT_ARN;

export const streamStarted = async (event) => {
  const client = new IvschatClient();

  const sendEventInput = {
    roomIdentifier: chatArn,
    eventName: 'STREAM_STARTED',
    attributes: {
      event: JSON.stringify(event),
    },
  }
  const sendEventRequest = new SendEventCommand(sendEventInput);
  const sendEventResponse = await client.send(sendEventRequest)

  const response = {
    statusCode: 200,
    headers: {
      'Access-Control-Allow-Origin': '*',
      'Access-Control-Allow-Methods': 'OPTIONS,GET,PUT,POST,DELETE',
      'Content-Type': 'application/json'
    },
    isBase64Encoded: false
  };
  response.statusCode = 200;
  response.body = JSON.stringify(sendEventResponse, '', 2);
  return response;
}; We'll need a file to create and deploy the function. In this template, we'll also add a definition for the EventBridge rule and create the necessary permissions for the rule to invoke the AWS Lambda function. Also note that I'm setting the as an environment variable so that it can be accessed in the AWS Lambda function. template.yaml DEMO_CHAT_ARN AWSTemplateFormatVersion: '2010-09-09'
Transform: 'AWS::Serverless-2016-10-31'
Description: Amazon IVS Functions
Globals:
  Function:
    Runtime: nodejs18.x
    Timeout: 30
    MemorySize: 128
  Api:
    Cors:
      AllowMethods: "'GET,POST,OPTIONS'"
      AllowHeaders: "'Content-Type'"
      AllowOrigin: "'*'"
      MaxAge: "'600'"
Resources:
  LambdaRefLayer:
    Type: AWS::Serverless::LayerVersion
    Properties:
      LayerName: sam-app-dependencies
      Description: Dependencies for sam app
      ContentUri: dependencies/
      CompatibleRuntimes:
        - nodejs18.x
      LicenseInfo: "MIT"
      RetentionPolicy: Retain

  IVSAccessPolicy:
    Type: AWS::IAM::Policy
    Properties:
      PolicyName: IVSAccess
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - 'ivschat:*'
            Resource: '*'
      Roles:
        - Ref: StreamStartedRole

  StreamStarted:
    Type: 'AWS::Serverless::Function'
    Properties:
      Environment:
        Variables:
          DEMO_CHAT_ARN: 'arn:aws:ivschat:us-east-1:[redacted]:room/[redacted]'
      Handler: index.streamStarted
      Layers:
        - !Ref LambdaRefLayer
      CodeUri: lambda/

  EventRule0:
    Type: AWS::Events::Rule
    Properties:
      Description: >-
        Rule to send a custom chat event when an Amazon IVS live stream session
        begins
      EventBusName: default
      EventPattern:
        source:
          - aws.ivs
        detail-type:
          - IVS Stream State Change
        detail:
          event_name:
            - Stream Start
      Name: demo-stream-started-0
      State: ENABLED
      Targets:
        -
          Arn: 
            Fn::GetAtt: 
              - "StreamStarted"
              - "Arn"
          Id: "StreamStartedTarget"

  PermissionForEventsToInvokeLambda: 
    Type: AWS::Lambda::Permission
    Properties: 
      FunctionName: 
        Ref: "StreamStarted"
      Action: "lambda:InvokeFunction"
      Principal: "events.amazonaws.com"
      SourceArn: 
        Fn::GetAtt: 
          - "EventRule0"
          - "Arn" Now we can this serverless application via the AWS SAM CLI. package If you're not familiar with AWS SAM, check out the 'getting started' . Note: guide sam package --template-file template.yaml --output-template-file packaged.yaml --s3-bucket ivs-demo And it. deploy sam deploy --template-file packaged.yaml --stack-name ivs-demo-stack --capabilities CAPABILITY_IAM Once deployed, you can verify that the EventBridge rule was created. One way to do that is to view your rules in the EventBridge console. You can also verify that the AWS Lambda was created via the console. Now we can start a broadcast on the Amazon IVS channel, and observe that the EventBridge rule has been invoked via the console. Listening For Custom Events on the Client Side Now that the we have custom events being published to an Amazon IVS chat room, we can set up our front-end to connect to the chat room and add some logic to our handler to play the stream when the custom event is received. The AWS Lambda function above sends the triggering event in the property of the custom event, so we can use the data from that triggering event to check the channel name and determine if we need to play the stream. The method is not defined in the example below, but it would contain any necessary logic to play the stream depending on your chosen video player. onmessage Attributes playStream() const connection = new WebSocket('[chat room endpoint]', '[chat token]');

connection.onmessage = (event) => {
  const data = JSON.parse(event.data);

  if (data.Type === 'MESSAGE') {
    // render chat message 
  }
  if (data.Type === 'EVENT') {
    switch (data.EventName) {
      case 'STREAM_STARTED':
        const triggerEvent = JSON.parse(data.Attributes.event);
        if (triggerEvent.detail.channel_name === 'demo-channel' && !isPlaying) {
          playStream();
        } 
        break;
    }
  }
} And that's it! Our stream will now automatically play when a broadcaster goes online. Summary In this post, we learned how to use EventBridge rules to send a custom Amazon IVS chat event when an Amazon IVS live stream session begins so that the live stream video can be automatically played. Question, comments, and feedback is always welcomed below. Also published . here

How to Autoplay Streams When You Go Live 

There are several options for broadcasting a live stream to an Amazon Interactive Video Service (Amazon IVS) channel. You (and your users) can choose one of the many third-party software options like OBS, Streamlabs Desktop, Lightstream, etc. You can also create a mobile experience with one of our . Another option for broadcasting is the which enables developers to create a dynamic, high-quality live stream broadcast experience directly in the browser. native broadcast SDKs for iOS or Android Web Broadcast SDK With the Web Broadcast SDK, you can provide streamers a custom interface to broadcast their webcam, microphone, and even share their desktop to an Amazon IVS channel. Sometimes streamers want the ability to include pre-recorded content in a live stream. Maybe they want to remind viewers of a highlight from a previous stream. Or, maybe they want to include their live reactions to a clip while it plays. Whatever the reason, it's easy to add pre-recorded video on demand (VOD) content to a live stream with the Web Broadcast SDK. Let's take a look. Adding VOD Content to a Live Stream I won't go over the basics of creating a broadcast with the Web Broadcast SDK in this post since I've already covered it in a on this blog. Instead, we'll focus on adding a pre-recorded video. To get started, we'll add a tag on the page that our broadcaster can use to play the VOD. previous post <video> <video id="vod-0" src="/video/vod-0.mp4" controls></video> In my demo app, I have included a few pieces of VOD content and an icon to allow the broadcaster to toggle back to 'camera only' mode. Your application might even include an upload feature to allow users to upload their own videos. Now that the assets are included in the front end, let's configure the page to broadcast this VOD to the live stream when it is played. <video> One approach is to automatically stream the VOD when the video is played by listening for the event. play const vod0 = document.getElementById('vod-0');
vod0.addEventListener('play', async (evt) => { 

}; Within the listener, we'll remove the user's webcam and microphone, and add the VOD (and the VOD audio) to the broadcast. The VOD is added via the method on the ( ) by passing the received in the event handler (the element). The method name ( ) might lead you to believe that only images can be added, but the docs clarify that the acceptable type for the source can be one of: , , , or . play addImageSource AmazonIVSBroadcastClient docs evt.target <video> addImageSource HTMLImageElement HTMLVideoElement HTMLCanvasElement ImageBitmap vod0.addEventListener('play', async (evt) => { 
  /// remove webcam
  const cameraExists = client.getVideoInputDevice('camera-0');
  if (cameraExists) await client.removeVideoInputDevice('camera-0');

  // remove user microphone
  const micExists = window.broadcastClient.getAudioInputDevice('mic-0');
  if (micExists) await window.broadcastClient.removeAudioInputDevice('mic-0');

  // add VOD to broadcast
  window.broadcastClient.addImageSource(
    evt.target, 
    'vod-0', 
    { index: 0 }
  );

  // add audio from VOD
  window.broadcastClient.addAudioInputDevice(
    vod0.captureStream(), 
    'vod-0-audio'
  );
}; Now, when the video is played, it will be exclusively streamed to the viewers on that channel without webcam or microphone inputs. We can listen for the event on the video to reset the webcam and microphone view so that broadcasters can pause to discuss a portion of the VOD. We can also listen for the event to reset the view when the VOD is finished playing. paused ended Adding a VOD with WebCam Overlay Adding an inline VOD that plays exclusively is a great feature, but sometimes broadcasters want to "react" during VOD content and include their webcam as an overlay to the pre-recorded content. Here we add the VOD as we did above, but also add a webcam stream on top of it by passing additional configuration properties ( , , , ) in the third argument to . height width x y addVideoInputDevice() vod1.addEventListener('play', async (evt) => {
  // remove webcams, VODs, desktop, etc...
  // [custom code here]

  // add VOD to broadcast (layer 0)
  window.broadcastClient.addImageSource(
    evt.target, 
    'vod-1', 
    { index: 0 }
  );

  // add camera as PIP
  const streamConfig = IVSBroadcastClient.STANDARD_LANDSCAPE;
  // get a webcam stream (getUserMedia())
  const videoStream = await getVideoStream(); 
  // add the webcam (layer 1 - on top of VOD)
  const preview = document.getElementById('broadcast-preview');
  window.broadcastClient.addVideoInputDevice(
    videoStream, 
    'pip-camera-0', 
    {
      index: 1,
      height: streamConfig.maxResolution.height * .25,
      width: streamConfig.maxResolution.width * .25,
      x: preview.width - preview.width / 4 - 20,
      y: preview.height - preview.height / 4 - 20
    }
  );

  // add audio from VOD
  window.broadcastClient.addAudioInputDevice(
    vod1.captureStream(), 
    'vod-0-audio'
  );
}); This gives us a nice picture-in-picture style interface. The broadcaster audio can be included so they can react/discuss the video. Again, you can utilize the and events as appropriate to handle state changes in your application. pause ended Summary In this post, we learned how to include pre-recorded VOD content in our Amazon IVS live streams with the Web Broadcast SDK. If you'd like to see the complete code behind this example, you can check it out . If you have any questions or suggestions for future posts that you'd like to see about Amazon IVS, please leave a comment below. here

How to Add Pre-Recorded Videos with the Amazon IVS Web Broadcast SDK

If you're planning on building a user-generated content (UGC) platform with Amazon Interactive Video Service (Amazon IVS), then you'll likely want to integrate some sort of dashboard to monitor the health of your user's live streams. As always, the provides a great overview of monitoring stream health. documentation In this post we'll go a bit further and show some examples of how to retrieve stream sessions, and retrieve health metrics for a live stream from Amazon CloudWatch. As an added bonus, we'll also look at generating a few useful charts that can be added to your UGC dashboard for realtime health monitoring. Amazon IVS stream health metrics are stored in Amazon CloudWatch. To retrieve these metrics with the AWS SDK for JavaScript v3, we can use the method from the Amazon CloudWatch client ( ). GetMetricDataCommand docs @aws-sdk/client-cloudwatch This method expects a few properties to filter the metrics to a specific time period and Amazon IVS channel as we'll see below. This post will focus on retrieving stream health metrics with the Amazon CloudWatch SDK, but vital stream health metrics can also be observed via Amazon EventBridge so you can take any necessary action when a stream becomes unhealthy or service limits are breached. Refer to the for more information. Note: documentation Retrieving Amazon IVS Stream Sessions Since we need a and to retrieve health metrics, it makes sense to retrieve a list of recent stream sessions via the Amazon IVS client ( ). To do this, we can use ( ) and pass it the for the channel that we're interested in. StartTime EndTime @aws-sdk/client-ivs ListStreamSessionsCommand docs ARN import { IvsClient, ListStreamSessionsCommand } from "@aws-sdk/client-ivs";

const ivsClient = new IvsClient();
const listStreamSessionsInput = {
  channelArn: process.env.DEMO_CHANNEL_ARN,
  maxResults: 3, // default=100
};
const listStreamSessionsRequest = new ListStreamSessionsCommand(listStreamSessionsInput);
const listStreamSessionsResponse = await ivsClient.send(listStreamSessionsRequest);

console.log(listStreamSessionsResponse) The response from will return an object. The key in that object contains an array of stream sessions, sorted by the most recent session. Active sessions are indicated by the absence of an . ListStreamSessionsCommand streamSessions endTime {
  "$metadata": {
    "httpStatusCode": 200,
    "requestId": "[redacted]",
    "cfId": "[redacted]",
    "attempts": 1,
    "totalRetryDelay": 0
  },
  "nextToken": "AQI...[redacted]...A==",
  "streamSessions": [
    {
      "endTime": undefined,
      "hasErrorEvent": false,
      "startTime": "2023-01-20T14:30:11.000Z",
      "streamId": "st-[redacted]"
    },
    {
      "endTime": "2023-01-19T16:12:37.000Z",
      "hasErrorEvent": false,
      "startTime": "2023-01-19T16:12:29.000Z",
      "streamId": "st-[redacted]"
    },
    {
      "endTime": "2023-01-19T16:12:25.000Z",
      "hasErrorEvent": false,
      "startTime": "2023-01-19T16:12:22.000Z",
      "streamId": "st-[redacted]"
    }
  ]
} Retrieving a Single Stream Session At this point, we have could use these timestamps to retrieve our stream metrics. But, it might be useful to grab a bit more detail such as the audio and video ingest configuration. To retrieve this info, we can use ( ). This method expects the , and a which we have already obtained in the above result. GetStreamSessionCommand docs ARN streamId import { IvsClient, GetStreamSessionCommand } from "@aws-sdk/client-ivs";
import util from "node:util";

const ivsClient = new IvsClient();
const getStreamSessionInput = {
  channelArn: process.env.DEMO_CHANNEL_ARN,
  streamId: 'st-[redacted]'
};
const getStreamSessionRequest = new GetStreamSessionCommand(getStreamSessionInput);
const getStreamSessionResponse = await ivsClient.send(getStreamSessionRequest);

console.log(
  util.inspect(getStreamSessionResponse, false, null, true)
); The method returns information about the session (in this case, an active live stream). Notice that contains some handy items like the codec, bitrate, framerate, etc. The object contains all of the events that have been triggered for this particular stream. GetStreamSessionCommand ingestConfiguration truncatedEvents {
  "$metadata": {
    "httpStatusCode": 200,
    "requestId": "[redacted]",
    "cfId": "[redacted]",
    "attempts": 1,
    "totalRetryDelay": 0
  },
  "streamSession": {
    "channel": {
      "arn": "[redacted]",
      "authorized": false,
      "ingestEndpoint": "[redacted]",
      "latencyMode": "LOW",
      "name": "demo-channel",
      "playbackUrl": "[redacted]",
      "recordingConfigurationArn": "[redacted]",
      "type": "STANDARD"
    },
    "ingestConfiguration": {
      "audio": {
        "channels": 2,
        "codec": "mp4a.40.2",
        "sampleRate": 48000,
        "targetBitrate": 128000
      },
      "video": {
        "avcLevel": "3.1",
        "avcProfile": "Baseline",
        "codec": "avc1.42C01F",
        "encoder": "",
        "targetBitrate": 8500000,
        "targetFramerate": 30,
        "videoHeight": 1080,
        "videoWidth": 1920
      }
    },
    "recordingConfiguration": {
      "arn": "[redacted]",
      "destinationConfiguration": {
        "s3": {
          "bucketName": "[redacted]"
        }
      },
      "state": "ACTIVE"
    },
    "startTime": "2023-01-20T14:30:11.000Z",
    "streamId": "st-[redacted]",
    "truncatedEvents": [
      {
        "eventTime": "2023-01-20T14:30:19.000Z",
        "name": "Recording Start",
        "type": "IVS Recording State Change"
      },
      {
        "eventTime": "2023-01-20T14:30:18.000Z",
        "name": "Stream Start",
        "type": "IVS Stream State Change"
      },
      {
        "eventTime": "2023-01-20T14:30:11.000Z",
        "name": "Session Created",
        "type": "IVS Stream State Change"
      }
    ]
  }
} Viewing Stream Health Metrics via the AWS Console If you're planning to integrate stream health metric data into your application, you will be using one of the AWS SDKs. But, if you're just looking to view health metric data on a less frequent basis, you can view them via the AWS Console. To view Amazon IVS related health metrics, select , then via the Amazon CloudWatch console. All metrics IVS We can browse metric data by choosing a dimension. For example, to view metrics by channel, select , and choose the desired time period, the channel, and the metric. By channel Retrieving Stream Health Metrics via the SDK We're ready to use the Amazon CloudWatch SDK to pull back health metric data for a channel based on the stream start and end times. expects a object ( ). GetMetricDataCommand GetMetricDataCommandInput docs As mentioned above, this object has and properties, as well as a property that should contain an array of queries depending on the health metrics that we'd like to retrieve. There are four properties related to Amazon IVS channel health that we're interested in here: , , , and . We'll construct an array for each metric, using as the and filtering on a specific channel by grabbing the portion of the channel ARN following . StartTime EndTime MetricDataQueries IngestAudioBitrate IngestVideoBitrate IngestFramerate KeyframeInterval AWS/IVS Namespace / import { CloudWatchClient, GetMetricDataCommand } from "@aws-sdk/client-cloudwatch";

const cloudWatchClient = new CloudWatchClient(); const getMetrics = async (arn, startTime, endTime) => {
  const streamHealthMetrics = [
    "IngestAudioBitrate",
    "IngestVideoBitrate",
    "IngestFramerate",
    "KeyframeInterval"
  ];
  const metricDataQueries = streamHealthMetrics.map((metric) => {
    return {
      Id: metric.toLowerCase(),
      MetricStat: {
        Metric: {
          MetricName: metric,
          Namespace: "AWS/IVS",
          Dimensions: [{ Name: "Channel", Value: arn.split("/")[1] }]
        },
        Period: 5,
        Stat: "Average",
      }
    }
  });

  const getMetricDataInput = {
    StartTime: startTime,
    EndTime: endTime,
    MetricDataQueries: metricDataQueries,
    MaxDatapoints: 100
  };

  const getMetricDataRequest = new GetMetricDataCommand(getMetricDataInput);
  const getMetricDataResponse = await cloudWatchClient.send(getMetricDataRequest);
  return getMetricDataResponse;
};

// get metrics for a session

const metrics = await getMetrics(
  process.env.DEMO_CHANNEL_ARN,
  new Date('2023-01-20T14:30:11.000Z'),
  new Date('2023-01-20T14:49:15.000Z')
); The result of calling will look similar to the following output: GetMetricDataCommand {
  "$metadata": {
    "httpStatusCode": 200,
    "requestId": "[redacted]",
    "attempts": 1,
    "totalRetryDelay": 0
  },
  "MetricDataResults": [
    {
      "Id": "ingestaudiobitrate",
      "Label": "IngestAudioBitrate",
      "Timestamps": [
        "2023-01-20T14:49:10.000Z"
      ],
      "Values": [
        31049.333057821852
      ],
      "StatusCode": "PartialData"
    },
    {
      "Id": "ingestvideobitrate",
      "Label": "IngestVideoBitrate",
      "Timestamps": [
        "2023-01-20T14:49:10.000Z"
      ],
      "Values": [
        3497988.4859657455
      ],
      "StatusCode": "PartialData"
    },
    {
      "Id": "ingestframerate",
      "Label": "IngestFramerate",
      "Timestamps": [
        "2023-01-20T14:49:10.000Z"
      ],
      "Values": [
        29.143738984724312
      ],
      "StatusCode": "PartialData"
    },
    {
      "Id": "keyframeinterval",
      "Label": "KeyframeInterval",
      "Timestamps": [
        "2023-01-20T14:49:10.000Z"
      ],
      "Values": [
        2.007629037
      ],
      "StatusCode": "PartialData"
    }
  ],
  "NextToken": "[redacted]",
  "Messages": []
} It's important to note that Amazon CloudWatch metrics are rolled up over time, so granular resolution decreases over time. 1-second metrics are available for 3 hours. 60-second metrics are available for 15 days. 5-minute metrics are available for 63 days. 1-hour metrics are available for 455 days (15 months). If we were to run the same query with a larger time range (depending on the availability of the data as mentioned above), we can filter and group the data based on a single metric and use that result to render a nice chart of the values. const videoBitrateMetrics = metrics
  .MetricDataResults
  .find((metric) => metric.Id === 'ingestvideobitrate');

const bitrateData = [];

videoBitrateMetrics.Timestamps
  .sort((a, b) => new Date(a) > new Date(b) ? 1 : -1)
  .forEach((t, i) => {
    bitrateData.push({
      timestamp: t,
      bitrate: videoBitrateMetrics.Values[i] / 1000,
    })
  });

console.log(JSON.stringify(bitrateData)); This produces an array of objects that looks like the following: [
  {
    "timestamp": "2023-01-20T14:47:05.000Z",
    "bitrate": 3497.9884859657454
  }
] We can use this filtered data to create some nice visualizations for our UGC dashboards. Some quick examples: Ingest Video Bitrate Ingest Audio Bitrate Ingest Framerate: Keyframe Interval: Generating Chart Images with the Amazon CloudWatch SDK Another super cool option is to generate the charts directly via the Amazon CloudWatch SDK. Check the for details. Here's an example of generating a chart for the metric via the Amazon CloudWatch SDK. documentation IngestFramerate const getMetricImage = async (arn, startDate, endDate) => {
  const cloudWatchClient = new CloudWatchClient();
  const getMetricWidgetImageInput = {
    MetricWidget: JSON.stringify({
      metrics: [
        [
          "AWS/IVS",
          "IngestFramerate",
          "Channel",
          arn.split("/")[1]
        ]
      ],
      start: startDate,
      end: endDate,
      period: 5
    })
  };
  const getMetricWidgetImageRequest = new GetMetricWidgetImageCommand(getMetricWidgetImageInput);
  const getMetricWidgetImageResponse = await cloudWatchClient.send(getMetricWidgetImageRequest);
  return getMetricWidgetImageResponse;
};

const metricImage = await getMetricImage(
  process.env.DEMO_CHANNEL_ARN,
  new Date('2023-01-20T14:30:11.000Z'),
  new Date('2023-01-20T14:49:15.000Z')
); The call to returns a JSON object that looks like the following: GetMetricWidgetImageCommand {
  '$metadata': {
    httpStatusCode: 200,
    requestId: '[redacted]',
    extendedRequestId: undefined,
    cfId: undefined,
    attempts: 1,
    totalRetryDelay: 0
  },
  MetricWidgetImage: Uint8Array(36660) [
    137, 80, 78,  
    ... 36560 more items
  ]
} We can convert the to a base64 string: Uint8Array const buffer = Buffer.from(metricImage.MetricWidgetImage);
console.log(buffer.toString('base64')); Which returns a base64 string: iVBORw0KGgoAAAANSUhEUgAAAlgAAAGQCAIAAAD9V4nPA... That can be used to render an image: We can weven combine multiple metrics in a single image by passing additional metrics to . GetMetricWidgetImageCommand Summary In this post, we looked at how to retrieve health metrics for an Amazon IVS live stream and render them as charts. Stay tuned for a future post where we dig into some of the other metrics available via Amazon CloudWatch related to live stream viewers.

How to Retrieve Health Metrics for an Amazon IVS Live Stream and Render Them as Charts

Cómo recuperar métricas de salud para una transmisión en vivo de Amazon IVS y representarlas como gráficos

Comment récupérer des mesures de santé pour un flux en direct Amazon IVS et les afficher sous forme de graphiques

Amazon IVS लाइव स्ट्रीम के लिए हेल्थ मेट्रिक्स कैसे प्राप्त करें और उन्हें चार्ट के रूप में प्रस्तुत करें

Amazon IVS ライブ ストリームのヘルス メトリクスを取得し、チャートとしてレンダリングする方法

Como recuperar métricas de integridade para uma transmissão ao vivo do Amazon IVS e renderizá-las como gráficos

Cách truy xuất số liệu sức khỏe cho luồng trực tiếp Amazon IVS và hiển thị chúng dưới dạng biểu đồ

如何检索 Amazon IVS 直播流的健康指标并将其呈现为图表

Most inter-systems communication components that use REST serialize their payload in JSON. As of now, JSON lacks a widely-used schema validation standard: is not widespread. Standard schema validation allows delegating the validation to a third-party library and being done with it. Without one, we must fall back to manual validation in the code. Worse, we must keep the validation code in sync with the schema. JSON Schema XML has schema validation out-of-the-box: an XML document can declare a grammar that it must conform to. SOAP, being based on XML, benefits from it too. Other serialization alternatives have a schema validation option: , , and . Interestingly enough, gRPC uses Protobuf to offer RPC across distributed components: e.g. Avro Kryo Protocol Buffers gRPC is a modern open source high performance Remote Procedure Call (RPC) framework that can run in any environment. It can efficiently connect services in and across data centers with pluggable support for load balancing, tracing, health checking and authentication. It is also applicable in last mile of distributed computing to connect devices, mobile applications and browsers to backend services. -- Why gRPC? Moreover, Protocol is a serialization mechanism, saving a lot of bandwidth. Thus, gRPC is an excellent option for inter-systems communication. But if all your components talk gRPC, how can simple clients call them? In this post, we will build a gRPC service and show how to call it from curl. binary A simple gRPC service The is exhaustive, so here's a summary: gRPC documentation gRPC is a Remote Procedure Call framework It works across a wide range of languages It relies on Protocol Buffers: Protocol buffers are Google’s language-neutral, platform-neutral, extensible mechanism for serializing structured data – think XML, but smaller, faster, and simpler. You define how you want your data to be structured once, then you can use special generated source code to easily write and read your structured data to and from a variety of data streams and using a variety of languages. -- Protocol Buffers It's part of the portfolio and is currently in the incubation stage CNCF Let's set up our gRPC service. We will use Java, Kotlin, Spring Boot, and a dedicated gRPC Spring Boot integration project. The project structure holds two projects, one for the model and one for the code. Let's start with the model project. I didn't want something complicated; reusing a simple example is enough: the request sends a string, and the response prefixes it with . We design this model in a dedicated Protobuf schema file: Hello syntax = "proto3";                                        //1

package ch.frankel.blog.grpc.model;                       //2

option java_multiple_files = true;                        //3
option java_package = "ch.frankel.blog.grpc.model";       //3
option java_outer_classname = "HelloProtos";              //3

service HelloService {                                    //4
    rpc SayHello (HelloRequest) returns (HelloResponse) {
    }
}

message HelloRequest {                                    //5
    string name = 1;                                      //6
}

message HelloResponse {                                   //7
    string message = 1;                                   //6
} Protobuf definition version Package Java-specific configuration Service definition Request definition Field definition. First comes the type, then the name, and finally, the order Response definition We shall use Maven to generate the Java boilerplate code: <project>
  <dependencies>
    <dependency>
      <groupId>io.grpc</groupId>                         
      <artifactId>grpc-stub</artifactId>
      <version>${grpc.version}</version>
    </dependency>
    <dependency>
      <groupId>io.grpc</groupId>                         
      <artifactId>grpc-protobuf</artifactId>
      <version>${grpc.version}</version>
    </dependency>
    <dependency>
      <groupId>jakarta.annotation</groupId>              
      <artifactId>jakarta.annotation-api</artifactId>
      <version>1.3.5</version>
      <optional>true</optional>
    </dependency>
  </dependencies>
  <build>
    <extensions>
      <extension>
        <groupId>kr.motd.maven</groupId>                 
        <artifactId>os-maven-plugin</artifactId>
        <version>1.7.1</version>
      </extension>
    </extensions>
    <plugins>
      <plugin>
        <groupId>org.xolstice.maven.plugins</groupId>    
        <artifactId>protobuf-maven-plugin</artifactId>
        <version>${protobuf-plugin.version}</version>
        <configuration>
          <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}</protocArtifact>
          <pluginId>grpc-java</pluginId>
          <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
        </configuration>
        <executions>
          <execution>
            <goals>
              <goal>compile</goal>
              <goal>compile-custom</goal>
            </goals>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</project> Compile-time dependencies Sniff information about the Operating System. Used in the next plugin Generate Java code from the file proto After compilation, the structure should look something like the following: We can package the classes in a JAR and use it in a web app project. The latter is in Kotlin, but only because it's my favourite JVM language. We only need a specific Spring Boot starter dependency to integrate gRPC endpoints with Spring Boot: <dependency>
  <groupId>net.devh</groupId>
  <artifactId>grpc-server-spring-boot-starter</artifactId>
  <version>2.14.0.RELEASE</version>
</dependency> Here's the significant bit: @GrpcService                                                        //1
class HelloService : HelloServiceImplBase() {                       //2
  override fun sayHello(
      request: HelloRequest,                                        //2
      observer: StreamObserver<HelloResponse>                       //3
  ) {
    with(observer) {
      val reply = HelloResponse.newBuilder()                        //2
                               .setMessage("Hello ${request.name}") //4
                               .build()
      onNext(reply)                                                 //5
      onCompleted()                                                 //5
    }
  }
} The detects the annotation and works its magic grpc-server-spring-boot-starter Reference classes generated in the above project The method signature allows a parameter. The class comes from StreamObserver grpc-stub.jar Get the request and prefix it to build the response message Play the events We can now start the web app with . ./mvnw spring-boot:run Testing the gRPC service The whole idea behind the post is that accessing the gRPC service with regular tools is impossible. To test, we need a dedicated tool nonetheless. I found . Let's install it and use it to list available services: grpcurl grpcurl --plaintext localhost:9090 list   #1-2 List all available gRPC services TLS verification without To avoid clashes between gRPC and other channels, , REST, Spring Boot uses another port e.g. ch.frankel.blog.grpc.model.HelloService   #1
grpc.health.v1.Health                     #2
grpc.reflection.v1alpha.ServerReflection  #2 The gRPC service we defined Two additional services provided by the custom starter We can also dive into the structure of the service: grpcurl --plaintext localhost:9090 describe ch.frankel.blog.grpc.model.HelloService service HelloService {
  rpc SayHello ( .ch.frankel.blog.grpc.model.HelloRequest ) returns ( .ch.frankel.blog.grpc.model.HelloResponse );
} Finally, we can call the service with data: grpcurl --plaintext -d '{"name": "John"}' localhost:9090 ch.frankel.blog.grpc.model.HelloService/SayHello {
  "message": "Hello John"
} Accessing the gRPC service with regular tools Imagine that we have a regular JavaScript client-side application that needs to access the gRPC service. What would be the alternatives? The general approach is through : grpc-web A JavaScript implementation of gRPC for browser clients. For more information, including a quick start, see the gRPC-web documentation. gRPC-web clients connect to gRPC services via a special proxy; by default, gRPC-web uses Envoy. In the future, we expect gRPC-web to be supported in language-specific web frameworks for languages such as Python, Java, and Node. For details, see the roadmap. -- grpc-web The description states a single limitation: it works only for JavaScript (as of now). However, there's another one. It's pretty intrusive. You need to get the file, generate boilerplate code, and make your code call it. You must do it for every client type. Worse, if the proto file changes, you need to regenerate the client code in each of them. proto An alternative exists, though, if you're using an API Gateway. I'll describe how to do it with , but perhaps other gateways can do the same. is a plugin that allows transcoding REST calls to gRPC and back again. Apache APISIX grpc-transcode The first step is to register the proto file in Apache APISIX: curl http://localhost:9180/apisix/admin/protos/1 -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d "{ \"content\": \"$(sed 's/"/\\"/g' ../model/src/main/proto/model.proto)\" }" The second step is to create a route with the above plugin: curl http://localhost:9180/apisix/admin/routes/1 -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d '
{
  "uri": "/helloservice/sayhello",                           #1
  "plugins": {
    "grpc-transcode": {
      "proto_id": "1",                                       #2
      "service": "ch.frankel.blog.grpc.model.HelloService",  #3
      "method": "SayHello"                                   #4
    }
  },
  "upstream": {
    "scheme": "grpc",
    "nodes": {
      "server:9090": 1
    }
  }
}' Define a granular route Reference the proto file defined in the previous command gRPC service gRPC method At this point, can make an HTTP request to the defined endpoint. Apache APISIX will transcode the call to gRPC, forward it to the defined service, get the response, and transcode it again. any client curl localhost:9080/helloservice/sayhello?name=John {"message":"Hello John"} Compared to , the API Gateway approach allows sharing the file with a single component: the Gateway itself. grpc-web proto Benefits of transcoding At this point, we can leverage the capabilities of the API Gateway. Imagine we want a default value if no is passed, , . Developers would happily set it in the code, but any change to the value would require a complete build and deployment. Changes can be nearly-instant if we put the default value in the Gateway's routes processing chain. Let's change our route accordingly: name e.g. World curl http://localhost:9180/apisix/admin/routes/1 -H 'X-API-KEY: edd1c9f034335f136f87ad84b625c8f1' -X PUT -d '
{
  "uri": "/helloservice/sayhello",
  "plugins": {
    "grpc-transcode": {
      ...
    },
    "serverless-pre-function": {                    #1
      "phase": "rewrite",                           #2
      "functions" : [
        "return function(conf, ctx)                 #3
          local core = require(\"apisix.core\")
          if not ngx.var.arg_name then
            local uri_args = core.request.get_uri_args(ctx)
            uri_args.name = \"World\"
            ngx.req.set_uri_args(uri_args)
          end
        end"
      ]
    }
  },
  "upstream": {
      ...
  }
}' Generic all-purpose plugin when none fits Rewrite the request Magic Lua code that does the trick Now, we can execute the request with an empty argument and get the expected result: curl localhost:9080/helloservice/sayhello?name {"message":"Hello World"} Conclusion In this post, we have briefly described gRPC and how it benefits inter-service communication. We developed a simple gRPC service using Spring Boot and . It comes at a cost, though: regular clients cannot access the service. We had to resort to to test it. The same goes for clients based on JavaScript - or the browser. grpc-server-spring-boot-starter grpcurl To bypass this limitation, we can leverage an API Gateway. I demoed how to configure Apache APISIX with the plugin to achieve the desired result. grpc-transcode The complete source code for this post can be found on . GitHub To go further: gRPC Protocol Buffers os-maven-plugin Maven Protocol Buffers Plugin gRPC-Spring-Boot-Starter grpcurl Apache APISIX grpc-transcode plugin Originally published at on March 16th, 2023 A Java Geek

Creating a Simple gRPC Service on the Client Side

There are several options for broadcasting a live stream to an Amazon Interactive Video Service (Amazon IVS) channel. You (and your users) can choose one of the many third-party software options like OBS, Streamlabs Desktop, Lightstream, etc. You can also create a mobile experience with one of our . Another option for broadcasting is the which enables developers to create a dynamic, high-quality live stream broadcast experience directly in the browser. With the Web Broadcast SDK, you can provide streamers a custom interface to broadcast their webcam, microphone, and even share their desktop to an Amazon IVS channel. native broadcast SDKs for iOS or Android Web Broadcast SDK Sometimes streamers want the ability to include pre-recorded content in a live stream. Maybe they want to remind viewers of a highlight from a previous stream. Or, maybe they want to include their live reactions to a clip while it plays. Whatever the reason, it's easy to add pre-recorded video on demand (VOD) content to a live stream with the Web Broadcast SDK. Let's take a look. Adding VOD Content to a Live Stream I won't go over the basics of creating a broadcast with the Web Broadcast SDK in this post since I've already covered it in a on this blog. Instead, we'll focus on adding a pre-recorded video. To get started, we'll add a tag on the page that our broadcaster can use to play the VOD. previous post <video> <video id="vod-0" src="/video/vod-0.mp4" controls></video> In my demo app, I have included a few pieces of VOD content and an icon to allow the broadcaster to toggle back to 'camera only' mode. Your application might even include an upload feature to allow users to upload their own videos. Now that the assets are included in the front end, let's configure the page to broadcast this VOD to the live stream when it is played. <video> One approach is to automatically stream the VOD when the video is played by listening for the event. play const vod0 = document.getElementById('vod-0');
vod0.addEventListener('play', async (evt) => { 

}; Within the listener, we'll remove the user's webcam and microphone, and add the VOD (and the VOD audio) to the broadcast. The VOD is added via the method on the ( ) by passing the received in the event handler (the element). The method name ( ) might lead you to believe that only images can be added, but the docs clarify that the acceptable type for the source can be one of: , , , or . play addImageSource AmazonIVSBroadcastClient docs evt.target <video> addImageSource HTMLImageElement HTMLVideoElement HTMLCanvasElement ImageBitmap vod0.addEventListener('play', async (evt) => { 
  /// remove webcam
  const cameraExists = client.getVideoInputDevice('camera-0');
  if (cameraExists) await client.removeVideoInputDevice('camera-0');

  // remove user microphone
  const micExists = window.broadcastClient.getAudioInputDevice('mic-0');
  if (micExists) await window.broadcastClient.removeAudioInputDevice('mic-0');

  // add VOD to broadcast
  window.broadcastClient.addImageSource(
    evt.target, 
    'vod-0', 
    { index: 0 }
  );

  // add audio from VOD
  window.broadcastClient.addAudioInputDevice(
    vod0.captureStream(), 
    'vod-0-audio'
  );
}; Now, when the video is played, it will be exclusively streamed to the viewers on that channel without webcam or microphone inputs. We can listen for the event on the video to reset the webcam and microphone view so that broadcasters can pause to discuss a portion of the VOD. We can also listen for the event to reset the view when the VOD is finished playing. paused ended Adding a VOD with WebCam Overlay Adding an inline VOD that plays exclusively is a great feature, but sometimes broadcasters want to "react" during VOD content and include their webcam as an overlay to the pre-recorded content. Here we add the VOD as we did above, but also add a webcam stream on top of it by passing additional configuration properties ( , , , ) in the third argument to . height width x y addVideoInputDevice() vod1.addEventListener('play', async (evt) => {
  // remove webcams, VODs, desktop, etc...
  // [custom code here]

  // add VOD to broadcast (layer 0)
  window.broadcastClient.addImageSource(
    evt.target, 
    'vod-1', 
    { index: 0 }
  );

  // add camera as PIP
  const streamConfig = IVSBroadcastClient.STANDARD_LANDSCAPE;
  // get a webcam stream (getUserMedia())
  const videoStream = await getVideoStream(); 
  // add the webcam (layer 1 - on top of VOD)
  const preview = document.getElementById('broadcast-preview');
  window.broadcastClient.addVideoInputDevice(
    videoStream, 
    'pip-camera-0', 
    {
      index: 1,
      height: streamConfig.maxResolution.height * .25,
      width: streamConfig.maxResolution.width * .25,
      x: preview.width - preview.width / 4 - 20,
      y: preview.height - preview.height / 4 - 20
    }
  );

  // add audio from VOD
  window.broadcastClient.addAudioInputDevice(
    vod1.captureStream(), 
    'vod-0-audio'
  );
}); This gives us a nice picture-in-picture style interface. The broadcaster audio can be included so they can react/discuss the video. Again, you can utilize the and events as appropriate to handle state changes in your application. pause ended Summary In this post, we learned how to include pre-recorded VOD content in our Amazon IVS live streams with the Web Broadcast SDK. If you'd like to see the complete code behind this example, you can check it out . If you have any questions or suggestions for future posts that you'd like to see about Amazon IVS, please leave a comment below. here

Adding Pre-Recorded Video to Live Streams with Amazon IVS Web Broadcast SDK

This post might not fall into the "normal" ways that you'd use live streaming with Amazon Interactive Video Service (Amazon IVS), but it's a fun project to play around and learn with and it illustrates some unique features of the Web Broadcast SDK (notably the ability to stream from a source other than a traditional camera and microphone). You may be familiar with the concept of "lofi radio" channels, but if not, they usually consist of some sort of animated character in a mostly static scene with a looped audio track containing lofi music. These live streams are popular among people looking for a non-distracting soundtrack to relax, study, or chat and make new friends. It doesn't fit the "traditional" user generated content (UGC) model of a broadcaster streaming a game or their webcam, but there is no arguing that these streams are hugely popular. For example, as of the time of publishing this article, the channel on YouTube currently has 11.7 million subscribers and boasts nearly 1.5 total views. Lofi Girl billion Maybe it's time that someone built an entire live streaming UGC platform that gives users the ability to create their own custom lofi channels and broadcast them to their friends? That's a free idea - run with it! I'm going to assume that you're new to Amazon IVS, so we'll walk through the process of creating your own lofi live stream from the beginning. If you're already familiar with Amazon IVS, feel free to jump past the parts that you're already comfortable with. Sign Up for a Free AWS Account Step one is to . New accounts will be eligible for 5 hours of basic input, 100 hours of SD output, and a generous amount of monthly chat messages for the first 12 months. This should be plenty of time to play around with Amazon IVS and learn all about how it works. sign up for a free AWS account Creating an Amazon IVS Channel I've blogged about how to , but the quickest way to create a one-off channel is via the AWS CLI ( ). The command to create a channel is: get started creating an Amazon IVS channel install docs $ aws ivs create-channel --name lofi-demo --latency-mode LOW --type BASIC This will return some important information about your stream: {
  "channel": {
    "arn": "arn:aws:ivs:us-east-1:[redacted]:channel/[redacted]",
    "authorized": false,
    "ingestEndpoint": "f99084460c35.global-contribute.live-video.net",
    "latencyMode": "LOW",
    "name": "lofi-demo",
    "playbackUrl": "https://f99084460c35.us-east-1.playback.live-video.net/api/video/v1/us-east-1.[redacted].channel.[redacted].m3u8",
    "recordingConfigurationArn": "",
    "tags": {},
    "type": "BASIC"
  },
  "streamKey": {
    "arn": "arn:aws:ivs:us-east-1:[redacted]:stream-key/[redacted]",
    "channelArn": "arn:aws:ivs:us-east-1:[redacted]:channel/[redacted]",
    "tags": {},
    "value": "sk_us-east-1[redacted]"
  }
} Keep this info handy, as we'll need the , , and in just a bit. ingestEndpoint playbackUrl streamKey Broadcasting to the Lofi Channel with the Web Broadcast SDK Now that we have a channel created, we can immediately start broadcasting to it. If you've done any live streaming, you're probably familiar with desktop streaming software (like OBS, Streamlabs Desktop, etc). Instead of using third-party software, we're going to write some code to broadcast our lofi stream directly from a browser. : You should have your own animation handy and an MP3 track. - please only utilize assets that you have obtained the proper licensing to use. If your animation is a GIF, you'll want to convert it to an MP4 file (there are plenty of tools available online for this). Before We Code Copyright is a thing The first step is to create an HTML page that includes the Web Broadcast SDK script, and contains a element for the broadcast preview, a element for the source animation, and a button to start the broadcast. The of the tag should point to your animation source and should have a width and height of (we're going to preview the video on the in just a bit, so no need for the source video to be visible). <canvas> <video> src <video> 1px <canvas> <!doctype html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta http-equiv="X-UA-Compatible" content="IE=edge">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>Lofi Radio</title>
  <script src="https://web-broadcast.live-video.net/1.2.0/amazon-ivs-web-broadcast.js"></script>
</head>
<body>

  <canvas id="broadcast-preview"></canvas>

  <video id="src-video" 
    src="/video/lofi.mp4" 
    style="width: 1px; height: 1px;"
    muted loop></video>

  <button id="broadcast-btn">Broadcast</button>

</body>
</html> Next, create an external JS file called (don't forget to include it in your HTML above). Start off this file with a listener and define an function to get things set up. In the function, we'll create an instance of the from the Web Broadcast SDK. We'll need to pass it the from our channel. We'll also create a click handler for the broadcast button that will call a function that we'll define in just a bit. lofi.js DOMContentLoaded init() init() IVSBroadcastClient ingestEndpoint toggleBroadcast() const init = () => {
  window.broadcastClient = IVSBroadcastClient.create({
    streamConfig: IVSBroadcastClient.BASIC_FULL_HD_LANDSCAPE,
    ingestEndpoint: [Your ingestEndpoint],
  });
  document.getElementById('broadcast-btn').addEventListener('click', toggleBroadcast);
};
document.addEventListener('DOMContentLoaded', init); To keep things simple, we're setting the into the window scope. In reality, you'll probably be using a framework so you'll usually avoid using the global scope like this. Note: broadcastClient window Next, we need to create our video stream and attach it to the . Modify the function as follows: broadcastClient init() const init = () => {
  window.broadcastClient = IVSBroadcastClient.create({
    streamConfig: IVSBroadcastClient.BASIC_FULL_HD_LANDSCAPE,
    ingestEndpoint: [Your ingestEndpoint],
  });

  window.video = document.getElementById('src-video');
  window.broadcastClient.addImageSource(window.video, 'video-track', { index: 0 });
  const preview = document.getElementById('broadcast-preview');
  window.broadcastClient.attachPreview(preview);

  document.getElementById('broadcast-btn').addEventListener('click', toggleBroadcast);
}; At this point, we've got our animation source created and added to the client. Next, let's add a function that will attach the MP3 audio source to the stream. createAudioStream() const createAudioStream = async () => {
  /* Music from Uppbeat (free for Creators!): https://uppbeat.io/t/vens-adams/alone-in-kyoto */
  const audioContext = new AudioContext();
  const mp3 = await fetch('/audio/alone-in-kyoto.mp3');
  const mp3Buffer = await mp3.arrayBuffer();
  const audioBuffer = await audioContext.decodeAudioData(mp3Buffer);
  const streamDestination = audioContext.createMediaStreamDestination();
  const bufferSource = audioContext.createBufferSource();
  bufferSource.buffer = audioBuffer;
  bufferSource.start(0);
  bufferSource.connect(streamDestination);
  bufferSource.loop = true; 
  window.broadcastClient.addAudioInputDevice(streamDestination.stream, 'audio-track');
}; And modify the function to add and call . init() async await createAudioStream() const init = async () => {
  window.broadcastClient = IVSBroadcastClient.create({
    streamConfig: IVSBroadcastClient.BASIC_FULL_HD_LANDSCAPE,
    ingestEndpoint: [Your ingestEndpoint],
  });

  window.video = document.getElementById('src-video');
  window.broadcastClient.addImageSource(window.video, 'video-track', { index: 0 });
  const preview = document.getElementById('broadcast-preview');
  window.broadcastClient.attachPreview(preview);

  await createAudioStream();

  document.getElementById('broadcast-btn').addEventListener('click', toggleBroadcast);
}; Finally, we can define the function to start/stop the stream when the button is clicked. toggleBroadcast() : you'll need to plug in your at this point. Treat this like any other sensitive credential and protect it from being committed to source control or included in a file that can be accessed by others. Note streamKey const toggleBroadcast = () => {
  if (!window.isBroadcasting) {
    window.video.play();
    window.broadcastClient
      .startBroadcast([your streamKey])
      .then(() => {
        window.isBroadcasting = true;
      })
      .catch((error) => {
        window.isBroadcasting = false;
        console.error(error);
      });
  }
  else {
    window.broadcastClient.stopBroadcast();
    window.video.pause();
    window.isBroadcasting = false;
  }
}; Now we can save and run the application. Here's how my page looks (with a bit of CSS applied for styling and layout). When I click the button, the animation begins to play and our live stream is broadcast to our channel. Broadcast : You will not be able to hear the channel audio from the broadcast tool. We'll look at playback in just a bit. Note Playing Back the Lofi Channel I've blogged about with Amazon IVS before, but we'll create a quick and simple player here to complete the demo. Create an HTML file, include the Amazon IVS player SDK, add a element, create an instance of the player (plug in your ), and initialize playback. live stream playback <video> playbackUrl <!doctype html>
<html lang="en">

<head>
  <meta charset="UTF-8">
  <meta http-equiv="X-UA-Compatible" content="IE=edge">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>Lofi Radio Playback</title>
  <script src="https://player.live-video.net/1.14.0/amazon-ivs-player.min.js"></script>
  <script>
    document.addEventListener('DOMContentLoaded', () => {
      const ivsPlayer = IVSPlayer.create();
      ivsPlayer.attachHTMLVideoElement(document.getElementById('video-player'));
      ivsPlayer.load([your playbackUrl]);
      ivsPlayer.play();
    })
  </script>
</head>

<body>
  <video id="video-player" controls autoplay playsinline></video>
</body>

</html> And now we can playback our lofi radio stream in it's own page! The nice thing about our code is that both the audio and animation will continue on a loop until your channel stops broadcasting. Bonus: Broadcasting Without a Browser Another option for broadcasting your own lofi stream is to avoid having any user interface and broadcast in headless mode with FFMPEG (more about that ). Here's an example command that would work for this channel. here $ ffmpeg \
    -re \
    -stream_loop -1 \
    -i ./public/video/lofi.mp4 \
    -stream_loop -1 \
    -i ./public/audio/alone-in-kyoto.mp3 \
    -map 0:v:0 \
    -map 1:a:0 \
    -c:v libx264 \
    -b:v 6000K \
    -maxrate 6000K \
    -pix_fmt yuv420p \
    -s 1920x1080 \
    -profile:v main \
    -preset veryfast \
    -force_key_frames "expr:gte(t,n_forced*2)" \
    -x264opts "nal-hrd=cbr:no-scenecut" \
    -acodec aac \
    -ab 160k \
    -ar 44100 \
    -f flv \
    $DEMO_STREAM_INGEST_ENDPOINT/$DEMO_STREAM_KEY Summary In this post, we learned how to create our own lofi radio live stream with Amazon IVS. To make your own lofi stream even more interactive, add with both and chat moderation. If you have any questions about this demo, please leave a comment below. live chat automated manual Also published . here

How to Build Your Own "LoFi Radio" Station and Stream it Live With Amazon IVS

1 Auto Recording Amazon IVS Live Streams to S3 2 Archiving Web Chat Messages with Amazon IVS Chat Logging 3 Amazon IVS Live Stream Playback with Chat Replay In our last few posts, we've looked at how to auto-record Amazon Interactive Video Service (Amazon IVS) live streams to Amazon S3, and how to log messages sent to an Amazon IVS chat room (see links above). In this post, we'll bring these two concepts together to complete the user experience and create on-demand playback of past live streams with full chat replay. As seen in our last post, chat messages that are logged to the logging destination include a GMT based timestamp representing the date and time that the event was posted by the Amazon IVS chat room. To accomplish a true on-demand replay of a live stream complete with interactive chat replay of a message at the closest moment in time possible, we need to obtain a regular stream of GMT based timestamps from the recorded stream that we can use to determine what chat messages should be visible at any given point in time in the stream playback. There's not currently a documented source that provides this information, but let's poke around with the Amazon IVS and see if we can find something that will help us out with this task. player SDK Amazon IVS Stream Metadata When trying to solve this problem, my first thought was to take a look at the metadata associated with a live stream to see if there is any valuable information hidden within it. Thankfully, there does appear to be a value in the regular stream of metadata that can be used for our chat playback purposes. In my testing, each stream contains ID3 metadata that appears to be injected by the Amazon IVS transcoding process. These ID3 tags contain a helpful timestamp that we can use to help with chat replay. To listen for these events, we can attach a handler that listens for the event type. This event type documented, but the don't say much about it or make any guarantees about what it may contain. IVSPlayer.MetadataEventType.ID3 is docs ? If you're concerned about using an undocumented feature, you could with the proper timestamp when new messages are posted into your Amazon IVS chat rooms. Keep in mind that there are events via the API. Want to Avoid Undocumented Features inject your own timed metadata into your live stream limits to the size and frequency of posting PutMetadata Listening for Metadata Events Let's set up an Amazon IVS player to playback a recorded stream using the Player SDK. First, we'll include the latest Amazon IVS player SDK via a tag. <script> Check out the blog series . If you have questions on getting started, post a comment on any post in that series (or below)! New to Amazon IVS? Getting Started with Amazon Interactive Video Service <script src="https://player.live-video.net/1.16.0/amazon-ivs-player.min.js"></script> As usual, we'll need to include a element in our HTML markup that will be used for playback. <video> <video id="video-player" muted controls autoplay playsinline></video> Now we can create an instance of the IVS player. I'm hardcoding the URL below, but you can obtain this URL via the method described in this post [todo: link]. const streamUrl = 'https://[redacted].cloudfront.net/ivs/v1/[redacted]/[redacted]/2022/11/17/18/6/[redacted]/media/hls/master.m3u8';
const videoEl = document.getElementById('video-player');
const ivsPlayer = IVSPlayer.create();
ivsPlayer.attachHTMLVideoElement(videoEl);
ivsPlayer.load(streamUrl);
ivsPlayer.play(); As mentioned above, to be useful for this purpose, we need a regular stream of timestamps. To figure out how often the ID3 metadata is received, let's add some timing. First, let's capture a timestamp as soon as the stream starts playing. ivsPlayer.addEventListener(IVSPlayer.PlayerState.PLAYING, (evt) => {
  window.time = Date.now();
}); Next, we'll add the ID3 event listener, log the timing, and reset the timer. ivsPlayer.addEventListener(IVSPlayer.MetadataEventType.ID3, (evt) => {
  const now = Date.now();
  console.log(`${(now - window.time) / 1000} seconds since last event`);
  window.time = now;
}); Now we can start playback and observe the console to see how often the events are fired. In my testing, the events are fired every 1-2 seconds. It's not realtime, but it's probably good enough for most scenarios. Now let's take a look at the event to see what data it contains. window.ivsPlayer.addEventListener(IVSPlayer.MetadataEventType.ID3, (evt) => {
  console.log(evt);
}); When we start playback with the listener above attached, we can see the following information logged to the browser console. This is very interesting info, but a bit cryptic. Based on my testing, seems to be timestamp that we're after. Let's modify the event handler to grab that timestamp and log it. transc_s window.ivsPlayer.addEventListener(IVSPlayer.MetadataEventType.ID3, (evt) => {
  const segmentMetadata = evt.find((tag) => tag.desc === 'segmentmetadata');
  const segmentMetadataInfo = JSON.parse(segmentMetadata.info[0]);
  const timestamp = segmentMetadataInfo['transc_s'];
  const timestampWithMs = timestamp * 1000;
  console.log(timestampWithMs);
  console.log(new Date(timestamp));
}); This produces the following output for my test. If I seek to a random moment in the video, the proper timestamp is always correct. This means that every 1-2 seconds we know a valid GMT timestamp of the moment in time that the event was capture in the stream. This means that we can assume that all chat messages that were sent prior to this timestamp have been posted to the chat and should be visible in the chat container. Retrieving the Chat Logs When my page loads, I can utilize the method outlined in the previous post [todo: link] in this series to retrieve the entire chat log for the stream and render it in the chat container . Since no messages should be visible at the very start of the stream, I'll make sure that they call contain a class that hides them from the user and store a data attribute with the proper timestamp so that I can know which messages should be visible given any timestamp in the stream. <div> window.chatLog = await getChatLogs(logGroupName, chatArn, startTime, endTime);
renderChat(); My function handles posting each message to the chat container. renderChat() const renderChat = () => {
  const chatContainer = document.getElementById('chat');
  window.chatLog.forEach(msg => {
    const msgTemplate = document.getElementById('chatMsgTemplate');
    const msgEl = msgTemplate.content.cloneNode(true);
    const ts = new Date(msg.event_timestamp).getTime() * 1000;
    msgEl.querySelector('.msg-container').setAttribute('data-timestamp', ts);
    msgEl.querySelector('.chat-username').innerHTML = msg.payload.Attributes.username;
    msgEl.querySelector('.msg').innerHTML = msg.payload.Content;
    chatContainer.appendChild(msgEl);
  });
}; Now I can modify the ID3 listener to call a function and pass it the current timestamp. replayChat() window.ivsPlayer.addEventListener(IVSPlayer.MetadataEventType.ID3, (evt) => {
  const segmentMetadata = evt.find((tag) => tag.desc === 'segmentmetadata');
  const segmentMetadataInfo = JSON.parse(segmentMetadata.info[0]);
  const timestamp = segmentMetadataInfo['transc_s'];
  const timestampWithMs = timestamp * 1000;
  replayChat(timestampWithMs);
}); In , I can find all of the chat nodes that contain a timestamp less than or equal to the current timestamp from the recorded stream and show/hide any chat message based on that timestamp. replayChat() const replayChat = (currentTimestamp) => {
  Array.from(document.querySelectorAll('[data-timestamp]')).forEach(node => {
    const chatMsgTs = Number(node.getAttribute('data-timestamp'));
    const isVisible = chatMsgTs <= currentTimestamp;
    if (isVisible) {
      node.classList.remove('d-none');
    }
    else {
      node.classList.add('d-none');
    }
  });
  const chatContainer = document.getElementById('chat');
  chatContainer.scrollTop = chatContainer.scrollHeight;
} At this point, we have achieved the goal of playing back a recorded Amazon IVS live stream with a full chat replay. Summary In this post, we looked at how to combine recorded Amazon IVS live streams with logged chat messages to create an on-demand replay of a stream with properly timed chat messages.

Stories, News & Mentions about STREAMHUB

Streamhub’s Stories on HackerNoon