A ConcurrentTask represents an asynchronous unit of work with the possibility of failure.

Underneath, each task represents a call to a JavaScript function and the runner handles batching and sequencing the calls.

Define a ConcurrentTask from an external JavaScript function with:

• The name of the registered function you want to call.
• What you expect to come back from the function when it returns.
• How to interpret errors coming from the function (exceptions or explicitly returned errors).
• The encoded args to pass to the function.

Say you wanted to interact with the node filesystem:

Define your task in Elm:

import ConcurrentTask exposing (ConcurrentTask)
import Json.Encode as Encode


type Error
    = Error String


readFile : String -> ConcurrentTask Error String
readFile path =
    ConcurrentTask.define
        { function = "fs:readFile"
        , expect = ConcurrentTask.expectString
        , errors = ConcurrentTask.expectThrows Error
        , args = Encode.object [ ( "path", Encode.string path ) ]
        }

And in your JavaScript runner:

import * as fs from "node:fs/promises"
import * as ConcurrentTask from "@andrewmacmurray/elm-concurrent-task";




const app = Elm.Main.init({});


ConcurrentTask.register({
  tasks: {
    "fs:readFile": (args) => fs.readFile(args.path),
  },
  ports: {
    send: app.ports.send,
    receive: app.ports.receive,
  },
});

A note on Errors:

The example fs:readFile Task has very simple error handling (turn any thrown exceptions into the Error type). This can be a great way to start, but what if you want more detailed errors?

The Errors section goes into more detail on different error handling strategies, including:

• how to define and handle more meaningful error types.
• bypassing all errors for Tasks which never fail (e.g. get the current time, log to the console).
• handling unexpected return data (e.g. the function returns an Int when you were expecting a String).

Decode the response of a JS function into an Elm value.

Run a JSON decoder on the response of a Task

Expect the response of a Task to be a String

Ignore the response of a Task

The Errors type provides different ways to capture errors for a ConcurrentTask.

Understanding Errors

ConcurrentTask has two main kinds of Errors:

Error

This is the x in the ConcurrentTask x a and represents an expected error as part of your task flow. You can handle these with mapError and onError.

UnexpectedError

You can think of these as unhandled errors that are not a normal part of your task flow.

The idea behind UnexpectedError is to keep your task flow types ConcurrentTask x a clean and meaningful, and optionally lift some of them into regular task flow where it makes sense. Two hooks you can use for this are onResponseDecoderFailure and onJsException.

See the section on UnexpectedErrors for more details.

The simplest Error handler. If a JS function throws an Exception, it will be wrapped in the provided Error type.

Maybe your JS function throws an AccessError:

import ConcurrentTask exposing (ConcurrentTask)


type Error
    = MyError String


example : ConcurrentTask Error String
example =
    ConcurrentTask.define
        { function = "functionThatThrows"
        , expect = ConcurrentTask.expectString
        , errors = ConcurrentTask.expectThrows MyError
        , args = Encode.null
        }

When the task is run it will complete with Task.Error (MyError "AccessError: access denied"). This can be transformed and chained using Task.mapError and Task.onError.

NOTE:

This kind of error handling can be useful to get started quickly, but it's often much more expressive and useful if you catch and explicitly return error data in your JS function that can be decoded with the expectError handler.

Decode explicit errors returned by a ConcurrentTask. Use this when you want more meaningful errors in your task.

This will decode the value from an error key returned by a JS function, e.g.:

return {
  error: {
    code: "MY_ERROR_CODE",
    message: "Something Went Wrong",
  }
}

IMPORTANT NOTES:

• If your function doesn't return an "error" key it will be interpreted as a success response.
• If your JS function throws an exception it will surface an UnhandledJsException - make sure to either:
  • catch these in your JS function and return them as structured errors.
  • catch them with the onJsException hook.
• If your error decoder fails the task will surface an ExpectErrorFailure.
  • This error is uncatchable, make sure to return data that matches your error decoder.

Maybe you want to handle different kinds of errors when writing to localStorage:

import ConcurrentTask exposing (ConcurrentTask)
import Json.Decode as Decode
import Json.Encode as Encode


type WriteError
    = QuotaExceeded
    | WriteBlocked


set : String -> String -> Task WriteError ()
set key value =
    ConcurrentTask.define
        { function = "storage:set"
        , expect = ConcurrentTask.expectWhatever
        , errors = ConcurrentTask.expectErrors decodeWriteError
        , args =
            Encode.object
                [ ( "key", Encode.string key )
                , ( "value", Encode.string value )
                ]
        }


decodeWriteError : Decode.Decoder WriteError
decodeWriteError =
    Decode.string
        |> Decode.andThen
            (\reason ->
                case reason of
                    "QUOTA_EXCEEDED" ->
                        Decode.succeed QuotaExceeded


                    "WRITE_BLOCKED" ->
                        Decode.succeed WriteBlocked


                    _ ->
                        Decode.fail ("Unknown WriteError Reason: " ++ reason)
            )

And on the JS side:

ConcurrentTask.register({
  tasks: {
    "storage:set": (args) => setItem(args),
  },
  ports: {
    send: app.ports.send,
    receive: app.ports.receive,
  },
});




function setItem(args) {
  try {
    localStorage.setItem(args.key, args.value);
  } catch (e) {
    if (e.name === "QuotaExceededError") {
      return {
        error: "QUOTA_EXCEEDED",
      };
    } else {
      return {
        error: "WRITE_BLOCKED",
      };
    }
  }
}

Only use this handler for functions that you don't expect to fail.

NOTE: If the decoder fails or the function throws an exception, these will be surfaced as UnexpectedErrors.

e.g. logging to the console:

import ConcurrentTask exposing (ConcurrentTask)


log : String -> ConcurrentTask x ()
log msg =
    ConcurrentTask.define
        { function = "console:log"
        , expect = ConcurrentTask.expectWhatever ()
        , errors = ConcurrentTask.expectNoErrors
        , args = Encode.string msg
        }

On the JS side:

ConcurrentTask.register({
  tasks: {
    "console:log": (msg) => console.log(msg),
  },
  ports: {
    send: app.ports.send,
    receive: app.ports.receive,
  },
});

Use this alongside other error handlers to lift a ResponseDecoderFailure's Json.Decode error into regular task flow.

Maybe you want to represent an unexpected response as a BadBody error for a http request:

import ConcurrentTask exposing (ConcurrentTask)


type Error
    = Timeout
    | NetworkError
    | BadStatus Int
    | BadUrl String
    | BadBody Decode.Error


request : Request a -> ConcurrentTask Error a
request options =
    ConcurrentTask.define
        { function = "http:request"
        , expect = ConcurrentTask.expectJson options.expect
        , errors = ConcurrentTask.expectErrors decodeHttpErrors
        , args = encodeArgs options
        }
        |> ConcurrentTask.onResponseDecoderFailure (BadBody >> Task.fail)

Use this to capture a raw JsException to lift it into the task flow.

NOTE: Tasks defined with expectThrows will never trigger this hook, make sure to only use it with expectErrors and expectNoErrors.

Transform the value of an Error (like map but for errors).

If the previous Task fails, catch that error and return a new Task (like andThen but for errors).

A Task that succeeds immediately when it's run.

A Task that fails immediately when it's run.

Chain the successful result of the previous Task into another one.

Maybe you want to do a timestamped Http request

import ConcurrentTask exposing (ConcurrentTask)
import ConcurrentTask.Http as Http
import ConcurrentTask.Time
import Time


task : ConcurrentTask Http.Error String
task =
    ConcurrentTask.Time.now
        |> Task.andThen (createArticle "my article")


createArticle : String -> Time.Posix -> ConcurrentTask Http.Error String
createArticle title time =
    Http.request
        { url = "http://blog.com/articles"
        , method = "POST"
        , headers = []
        , expect = Http.expectString
        , body = Http.jsonBody (encodeArticle title time)
        }

Create a Task from a Result error value. The task will either immediately succeed or fail when run.

Maybe you want to chain together tasks with CSV parsing:

import ConcurrentTask exposing (ConcurrentTask)
import Csv


task : ConcurrentTask Error CsvData
task =
    readFile |> ConcurrentTask.andThen parseCsv


parseCsv : String -> ConcurrentTask Error CsvData
parseCsv raw =
    Csv.decode decoder raw
        |> ConcurrentTask.fromResult
        |> ConcurrentTask.mapError CsvError

Similar to andThen but ignores the successful result of the previous Task.

Maybe you want to save a file then log a message to the console:

import ConcurrentTask exposing (ConcurrentTask)


task : ConcurrentTask Error ()
task =
    saveFile |> ConcurrentTask.andThenDo (log "file saved")

Succeed with a hardcoded value after the previous Task

Maybe you want to do some Tasks on a User but allow it to be chained onwards:

import ConcurrentTask exposing (ConcurrentTask)


saveUser : User -> ConcurrentTask Error User
saveUser user =
    saveToDatabase user
        |> ConcurrentTask.andThenDo (log "user saved")
        |> ConcurrentTask.return user

Perform a List of tasks concurrently (similar to Promise.all() in JavaScript) and return the results in a List.

If any of the subtasks fail the whole ConcurrentTask will fail.

Perform a List of tasks one after the other and return the results in a List.

If any of the subtasks fail the whole ConcurrentTask will fail.

Transform the value from a task.

Maybe you want to find what time it is in one hour.

import ConcurrentTask as exposing (ConcurrentTask)
import ConcurrentTask.Time
import Time


timeInOneHour : ConcurrentTask x Time.Posix
timeInOneHour =
    ConcurrentTask.map addOneHour ConcurrentTask.Time.now


addOneHour : Time.Posix -> Time.Posix
addOneHour time =
    Time.millisToPosix (Time.posixToMillis time + 60 * 60 * 1000)

Combine an arbitrary number of tasks together concurrently.

Maybe you want to load multiple pieces of config into a record:

import ConcurrentTask exposing (ConcurrentTask)


type alias Config =
    { dbConfig : DbConfig
    , yamlConfig : YamlConfig
    , envFile : EnvFile
    }


 loadConfig : ConcurrentTask Error Config
 loadConfig =
    ConcurrentTask.succeed Config
        |> ConcurrentTask.andMap loadDbConfig
        |> ConcurrentTask.andMap loadYamlConfig
        |> ConcurrentTask.andMap loadEnvFile

Run two tasks concurrently and combine their results.

import ConcurrentTask exposing (ConcurrentTask)
import ConcurrentTask.Time
import Time


loadUserAndTime : ConcurrentTask Error ( User, Time.Posix )
loadUserAndTime =
    ConcurrentTask.map2 Tuple.pair loadUser ConcurrentTask.Time.now

Run three tasks concurrently and combine their results.

Run four tasks concurrently and combine their results.

Run five tasks concurrently and combine their results.

Start a ConcurrentTask. This needs:

• A task Pool (The internal model to keep track of task progress).
• The send port.
• The Msg to be called when the task completes.
• Your ConcurrentTask to be run.

Make sure to update your Model and pass in the Cmd returned from attempt. e.g. in a branch of update:

let
    ( tasks, cmd ) =
        ConcurrentTask.attempt
            { send = send
            , pool = model.pool
            , onComplete = OnComplete
            }
            myTask
in
( { model | tasks = tasks }, cmd )

The value returned from a task when it completes (returned in the OnComplete msg).

Can be either:

• Success a - the task succeeded with no errors, woo!
• Error x - the task failed with an expected error.
• UnexpectedError - the task failed with an unexpected error (see the section on Error Handling for more details).

This error will surface if something unexpected has happened during the task flow.

Catchable Errors

These errors will be surfaced if not handled during task flow:

• UnhandledJsException - a task threw an exception and was not caught with an error handler (can be caught with expectThrows and onJsException).
• ResponseDecoderFailure - a task returned an unexpected response (can be caught with onResponseDecoderFailure).

Uncatchable Errors

These errors will always surface, as they are assumed to have no meaningful way to recover from during regular task flow:

• ErrorsDecoderFailure - a task returned error data in an unexpected format when using an expectErrors handler.
• MissingFunction - a task tried to call a function in the JS runner which was not registered.
• InternalError - something went wrong with the runner internals - this should not happen, but if you see this error please leave details and an issue.

Subscribe to updates from the JavaScript task runner.

This needs:

• The send port.
• The receive port.
• The Msg to be called with the updated progress.
• The task Pool stored in your model.

You can wire this in like so:

subscriptions : Model -> Sub Msg
subscriptions model =
    ConcurrentTask.onProgress
        { send = send
        , receive = receive
        , onProgress = OnProgress
        }
        model.tasks

Make sure to update your Model and pass in the Cmd in your OnProgress branch in update:

OnProgress ( tasks, cmd ) ->
    ( { model | tasks = tasks }, cmd )

A Pool keeps track of each task's progress, and allows multiple Task attempts to be in-flight at the same time.

Create an empty ConcurrentTask Pool.

Right now it doesn't expose any functionality, but it could be used in the future to do things like:

• Buffer the number of in-flight tasks (e.g. a server request queue, or database connection pool).
• Handle graceful process termination (e.g. abort or cleanup all in-flight tasks).
• Expose metrics on previous or running tasks.