Access Data Streams Using Automation

Read data from a Data Stream and validate the answer on-chain. This example uses a Chainlink Automation Log Trigger to check for events that require data. For this example, the log trigger comes from a simple emitter contract. Chainlink Automation then uses StreamsLookup to retrieve a signed report from the Data Streams Engine, return the data in a callback, and run the performUpkeep function on your registered upkeep contract. The performUpkeep function calls the verify function on the verifier contract.

Before you begin

Data Streams are currently available only in Early Access. Upkeeps for Data Streams will not run without first acquiring access. Contact us to talk to an expert before you deploy the example contracts in this tutorial.
If you are new to smart contract development, learn how to Deploy Your First Smart Contract so you are familiar with the tools that are necessary for this guide:
- The Solidity programming language
- The MetaMask wallet
- The Remix development environment
Acquire testnet funds. This guide requires a testnet ETH on Arbitrum Sepolia.
- Use the Arbitrum Bridge to transfer testnet ETH from Ethereum Sepolia to Arbitrum Sepolia. Testnet ETH is available at one of several faucets.
- Testnet LINK is available for Arbitrum Sepolia at faucets.chain.link.
Learn how to Fund your contract with LINK.

Deploy the Chainlink Automation upkeep

Deploy an upkeep that is enabled to retrieve data from Data Streams. For this example, you will read from the ETH/USD stream with ID 0x00027bbaff688c906a3e20a34fe951715d1018d262a5b66e38eda027a674cd1b on Arbitrum Sepolia. For a complete list of available assets, IDs, and verifier proxy addresses, see the Stream Identifiers page.

Select the Arbitrum Sepolia network in MetaMask.
Open the example upkeep in Remix.

Open in Remix What is Remix?
In the Solidity Compiler tab, select the 0.8.16 Solidity compiler and the StreamsUpkeep contract.
Compile the contract. You can ignore the warning messages for this example.
In the Deploy & Run tab, select Injected Provider as your Environment. For this example, you should be using Arbitrum Sepolia.
Deploy the contract with the following constructor variables:
- FeeAddress (WETH): 0xe39Ab88f8A4777030A534146A9Ca3B52bd5D43A3
- VerifierAddress: 0x2ff010DEbC1297f19579B4246cad07bd24F2488A
Record the contract address.

Deploy the emitter

This contract emits logs that trigger the upkeep. This code can be part of your dApp. For example, you might emit log triggers when your users initiate a trade or other action that requires data retrieval. For this Getting Started guide, we will use a very simple emitter so you can test the upkeep and data retrieval.

Select the Arbitrum Sepolia network in MetaMask.
Open the example emitter in Remix.

Open in Remix What is Remix?
Compile the contract.
In the Deploy & Run tab, select Injected Provider as your Environment.
Deploy the contract. Record the contract address.

Register the upkeep

Register a new Log Trigger upkeep. See Automation Log Triggers to learn more about how to register Log Trigger upkeeps.

Go to the Chainlink Automation UI for Arbitrum Sepolia and connect your browser wallet.
Click Register new Upkeep.
Select the Log Trigger upkeep type and click Next.
Specify a name for the upkeep.
Specify the upkeep contract address that you saved earlier and click Next.
Specify the emitter contract address that you saved earlier. This tells Chainlink Automation what contracts to watch for log triggers.
Specify a Starting balance of 1 testnet LINK for this example. You can retrieve unused LINK later.
Leave the Check data value and other fields blank for now, and click Register Upkeep. MetaMask prompts you to confirm the transaction.

Fund the upkeep contract

In this example, the upkeep contract pays for the verification. The Automation subscription does not cover that cost. Send LINK to your upkeep contract to pay for the verification costs.

In MetaMask, send 1 testnet LINK to your contract address before you proceed to the next step and trigger the upkeep.

Emit a log

Now you can use your emitter contract to emit a log and initiate the upkeep, which retrieves data for the specified Data Streams asset ID.

Go to sepolia.arbiscan.io.
Under the Deployed Contracts list, click the emitLog button to run the function and emit a log. MetaMask prompts you to accept the transaction.

After the transaction is complete, the log is emitted and the upkeep is triggered. You can find the upkeep transaction hash at Chainlink Automation UI. Check to make sure the transaction is successful.

View the retrieved price

The retrieved price is stored as a variable in the contract and is also emitted in the logs.

In Remix, go to the Deploy & Run tab.
Under the Deployed Contracts, find the deployed upkeep contract and view the variables.
Click the last_retrieved_price variable to view the retrieved price.

Alternatively, you can view the price emitted in the logs for your upkeep transaction. You can find the upkeep transaction hash at Chainlink Automation UI and view the logs in the explorer.

Examine the code

The example code that you deployed has all of the interfaces and functions required to work with Chainlink Automation as an upkeep contract. It follows a similar flow to the trading flow in the Architecture documentation, but uses a basic log emitter to simulate the client contract that would initiate a StreamsLookup. After the contract receives and verifies the report, performUpkeep emits a PriceUpdate log message with the price. You could modify this to use the data in a way that works for your specific use case and application.

The code example uses revert with StreamsLookup to convey call information about what streams to retrieve. See the eip-3668 rationale for more information on the use of revert in this way.

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.16;

import {Common} from "@chainlink/contracts/src/v0.8/libraries/Common.sol";
import {StreamsLookupCompatibleInterface} from "@chainlink/contracts/src/v0.8/automation/interfaces/StreamsLookupCompatibleInterface.sol";
import {ILogAutomation, Log} from "@chainlink/contracts/src/v0.8/automation/interfaces/ILogAutomation.sol";
import {IRewardManager} from "@chainlink/contracts/src/v0.8/llo-feeds/interfaces/IRewardManager.sol";
import {IVerifierFeeManager} from "@chainlink/contracts/src/v0.8/llo-feeds/interfaces/IVerifierFeeManager.sol";
import {IERC20} from "@chainlink/contracts/src/v0.8/vendor/openzeppelin-solidity/v4.8.0/contracts/interfaces/IERC20.sol";

/**
 * THIS IS AN EXAMPLE CONTRACT THAT USES UN-AUDITED CODE FOR DEMONSTRATION PURPOSES.
 * DO NOT USE THIS CODE IN PRODUCTION.
 */

// Custom interfaces for IVerifierProxy and IFeeManager
interface IVerifierProxy {
    function verify(
        bytes calldata payload,
        bytes calldata parameterPayload
    ) external payable returns (bytes memory verifierResponse);

    function s_feeManager() external view returns (IVerifierFeeManager);
}

interface IFeeManager {
    function getFeeAndReward(
        address subscriber,
        bytes memory unverifiedReport,
        address quoteAddress
    ) external returns (Common.Asset memory, Common.Asset memory, uint256);

    function i_linkAddress() external view returns (address);

    function i_nativeAddress() external view returns (address);

    function i_rewardManager() external view returns (address);
}

contract StreamsUpkeep is ILogAutomation, StreamsLookupCompatibleInterface {
    struct BasicReport {
        bytes32 feedId; // The feed ID the report has data for
        uint32 validFromTimestamp; // Earliest timestamp for which price is applicable
        uint32 observationsTimestamp; // Latest timestamp for which price is applicable
        uint192 nativeFee; // Base cost to validate a transaction using the report, denominated in the chain’s native token (WETH/ETH)
        uint192 linkFee; // Base cost to validate a transaction using the report, denominated in LINK
        uint32 expiresAt; // Latest timestamp where the report can be verified on-chain
        int192 price; // DON consensus median price, carried to 8 decimal places
    }

    struct PremiumReport {
        bytes32 feedId; // The feed ID the report has data for
        uint32 validFromTimestamp; // Earliest timestamp for which price is applicable
        uint32 observationsTimestamp; // Latest timestamp for which price is applicable
        uint192 nativeFee; // Base cost to validate a transaction using the report, denominated in the chain’s native token (WETH/ETH)
        uint192 linkFee; // Base cost to validate a transaction using the report, denominated in LINK
        uint32 expiresAt; // Latest timestamp where the report can be verified on-chain
        int192 price; // DON consensus median price, carried to 8 decimal places
        int192 bid; // Simulated price impact of a buy order up to the X% depth of liquidity utilisation
        int192 ask; // Simulated price impact of a sell order up to the X% depth of liquidity utilisation
    }

    struct Quote {
        address quoteAddress;
    }

    event PriceUpdate(int192 indexed price);

    IVerifierProxy public verifier;

    address public FEE_ADDRESS;
    string public constant DATASTREAMS_FEEDLABEL = "feedIDs";
    string public constant DATASTREAMS_QUERYLABEL = "timestamp";
    int192 public last_retrieved_price;

    // This example reads the ID for the basic ETH/USD price report on Arbitrum Sepolia.
    // Find a complete list of IDs at https://docs.chain.link/data-streams/stream-ids
    string[] public feedIds = [
        "0x00027bbaff688c906a3e20a34fe951715d1018d262a5b66e38eda027a674cd1b"
    ];

    constructor(address _verifier) {
        verifier = IVerifierProxy(_verifier);
    }

    // This function uses revert to convey call information.
    // See https://eips.ethereum.org/EIPS/eip-3668#rationale for details.
    function checkLog(
        Log calldata log,
        bytes memory
    ) external returns (bool upkeepNeeded, bytes memory performData) {
        revert StreamsLookup(
            DATASTREAMS_FEEDLABEL,
            feedIds,
            DATASTREAMS_QUERYLABEL,
            log.timestamp,
            ""
        );
    }

    // The Data Streams report bytes is passed here.
    // extraData is context data from feed lookup process.
    // Your contract may include logic to further process this data.
    // This method is intended only to be simulated off-chain by Automation.
    // The data returned will then be passed by Automation into performUpkeep
    function checkCallback(
        bytes[] calldata values,
        bytes calldata extraData
    ) external pure returns (bool, bytes memory) {
        return (true, abi.encode(values, extraData));
    }

    // function will be performed on-chain
    function performUpkeep(bytes calldata performData) external {
        // Decode the performData bytes passed in by CL Automation.
        // This contains the data returned by your implementation in checkCallback().
        (bytes[] memory signedReports, bytes memory extraData) = abi.decode(
            performData,
            (bytes[], bytes)
        );

        bytes memory unverifiedReport = signedReports[0];

        (, /* bytes32[3] reportContextData */ bytes memory reportData) = abi
            .decode(unverifiedReport, (bytes32[3], bytes));

        // Report verification fees
        IFeeManager feeManager = IFeeManager(address(verifier.s_feeManager()));
        IRewardManager rewardManager = IRewardManager(
            address(feeManager.i_rewardManager())
        );

        address feeTokenAddress = feeManager.i_linkAddress();
        (Common.Asset memory fee, , ) = feeManager.getFeeAndReward(
            address(this),
            reportData,
            feeTokenAddress
        );

        // Approve rewardManager to spend this contract's balance in fees
        IERC20(feeTokenAddress).approve(address(rewardManager), fee.amount);

        // Verify the report
        bytes memory verifiedReportData = verifier.verify(
            unverifiedReport,
            abi.encode(feeTokenAddress)
        );

        // Decode verified report data into BasicReport struct
        BasicReport memory verifiedReport = abi.decode(
            verifiedReportData,
            (BasicReport)
        );

        // Log price from report
        emit PriceUpdate(verifiedReport.price);

        // Store the price from the report
        last_retrieved_price = verifiedReport.price;
    }

    fallback() external payable {}
}

Open in Remix What is Remix?

Debugging StreamsLookup

Read our debugging section to learn how to identify and resolve common errors when using StreamsLookup.

Access Data Streams Using Automation

What's next

Stay updated on the latest Chainlink news