Rails + OpenAI Realtime API

application.js

/* eslint no-console:0 */

import "@hotwired/turbo-rails";
import SiriWave from "siriwave";

require("@rails/activestorage").start();
require("local-time").start();

import "./channels";
import "./controllers";
import "./src/**/*";


var siriWave = new SiriWave({
  container: document.getElementById("siri-container"),
  style: "ios",
  color: "#4ade80",
  cover: true,
  height: 200,
  speed: 0.3,
});

siriWave.setAmplitude(0);
siriWave.start();

document.addEventListener("turbo:load", () => {
  let audioContext;
  let audioBuffer = new Float32Array(0); // Continuous buffer to store audio samples
  let isPlaying = false;
  let scriptProcessor;
  let analyserNode;
  let amplitudeData;

  function initAudioContext() {
    if (!audioContext) {
      audioContext = new (window.AudioContext || window.webkitAudioContext)({
        sampleRate: 24000, // Match the sample rate to the incoming audio
      });
      console.log(
        "AudioContext initialized with sample rate:",
        audioContext.sampleRate,
      );

      // Create a ScriptProcessorNode for real-time audio playback
      scriptProcessor = audioContext.createScriptProcessor(4096, 1, 1);
      scriptProcessor.connect(audioContext.destination);
      scriptProcessor.onaudioprocess = handleAudioProcess;

      // Create an AnalyserNode to analyse the audio signal
      analyserNode = audioContext.createAnalyser();
      analyserNode.fftSize = 256; // Controls the granularity of the analysis
      amplitudeData = new Uint8Array(analyserNode.frequencyBinCount);

      // Connect the analyser to the script processor
      scriptProcessor.connect(analyserNode);
      analyserNode.connect(audioContext.destination);
    }
  }

  function handleAudioProcess(event) {
    const outputBuffer = event.outputBuffer.getChannelData(0);
    const bufferLength = outputBuffer.length;

    if (audioBuffer.length >= bufferLength) {
      // Copy data from the audioBuffer to the output buffer
      outputBuffer.set(audioBuffer.subarray(0, bufferLength));

      // Remove the played samples from the audioBuffer
      audioBuffer = audioBuffer.subarray(bufferLength);
    } else {
      // If there is not enough audio left, fill with silence
      outputBuffer.set(audioBuffer);
      audioBuffer = new Float32Array(0); // Clear the buffer once all audio has been played
    }

    if (audioBuffer.length === 0) {
      isPlaying = false;
    }

    // Get the amplitude data
    analyserNode.getByteTimeDomainData(amplitudeData);

    // Calculate the average amplitude (RMS)
    let sum = 0;
    for (let i = 0; i < amplitudeData.length; i++) {
      const value = (amplitudeData[i] - 128) / 128; // Normalize the data to [-1, 1]
      sum += value * value;
    }
    const rms = Math.sqrt(sum / amplitudeData.length);
    console.log("Current amplitude (RMS):", rms);
    siriWave.setAmplitude(rms * 20.0);
  }

  window.playAudioDelta = function (audioData) {
    console.log("Received audio data length:", audioData.length);

    if (!audioData || typeof audioData !== "string") {
      console.error("Invalid audioData:", audioData);
      return;
    }

    try {
      initAudioContext();

      let base64String = audioData.replace(/-/g, "+").replace(/_/g, "/");
      while (base64String.length % 4 !== 0) {
        base64String += "=";
      }
      base64String = base64String.replace(/[^A-Za-z0-9+/=]/g, "");

      const binaryString = atob(base64String);
      const bytes = new Uint8Array(binaryString.length);
      for (let i = 0; i < binaryString.length; i++) {
        bytes[i] = binaryString.charCodeAt(i);
      }

      const buffer = bytes.buffer;
      const numOfSamples = Math.floor(buffer.byteLength / 2); // Ensure integer number of samples
      const dataView = new DataView(buffer);
      const newSamples = new Float32Array(numOfSamples);

      for (let i = 0; i < numOfSamples; i++) {
        const sample = dataView.getInt16(i * 2, true);
        newSamples[i] = sample / 32768; // Normalize 16-bit PCM to float [-1.0, 1.0]
      }

      // Append the new samples to the existing audio buffer
      const updatedBuffer = new Float32Array(
        audioBuffer.length + newSamples.length,
      );
      updatedBuffer.set(audioBuffer);
      updatedBuffer.set(newSamples, audioBuffer.length);
      audioBuffer = updatedBuffer;

      console.log(
        "Audio chunk added to buffer. Buffer length (samples):",
        audioBuffer.length,
      );

      if (!isPlaying) {
        isPlaying = true;
        console.log("Starting playback");
      }
    } catch (error) {
      console.error("Error processing audio data:", error);
    }
  };

  window.stopAudioPlayback = function () {
    console.log("Stopping audio playback");
    isPlaying = false;
    audioBuffer = new Float32Array(0); // Clear buffer
    if (audioContext) {
      audioContext.close().then(() => {
        audioContext = null;
        console.log("AudioContext closed and playback stopped.");
      });
    }
  };

  initAudioContext();
  console.log("Audio playback system initialized");
});

function handleWebSocketMessage(data) {
  console.log("Received WebSocket message:", data.type);
  if (data.type === "input_audio_buffer.speech_started") {
    window.stopAudioPlayback();
  } else if (data.type === "audio") {
    window.playAudioDelta(data.data);
  }
}

open_ai_channel.js

// app/javascript/channels/open_ai_channel.js

import consumer from "./consumer";

let subscription = null;

document.addEventListener("DOMContentLoaded", () => {
  subscription = consumer.subscriptions.create("OpenAiChannel", {
    connected() {
      console.log("Connected to OpenAI channel");
    },

    disconnected() {
      console.log("Disconnected from OpenAI channel");
    },

    sendAudio(audioData) {
      this.perform("append_audio", { type: "audio", audio: audioData });
    },

    closeSession() {
      this.perform("close_session");
    },

    restartSession() {
      this.perform("restart_session");
    },
    received(data) {
      // Called when there's incoming data on the websocket for this channel
      console.log("Received data:", data);
      if (data.type === "audio") {
        playAudioDelta(data.data);
      }

      if (data.type === "input_audio_buffer.speech_started") {
        console.log("Speech started");
        this.perform("cancel_response");
        window.stopAudioPlayback();
      }
    },
  });

  // Button elements
  const startButton = document.getElementById("start-recording");
  const stopButton = document.getElementById("stop-recording");

  let audioContext;

  // Converts Float32Array of audio data to PCM16 ArrayBuffer
  function floatTo16BitPCM(float32Array) {
    const buffer = new ArrayBuffer(float32Array.length * 2);
    const view = new DataView(buffer);
    let offset = 0;
    for (let i = 0; i < float32Array.length; i++, offset += 2) {
      let s = Math.max(-1, Math.min(1, float32Array[i]));
      view.setInt16(offset, s < 0 ? s * 0x8000 : s * 0x7fff, true);
    }
    return buffer;
  }

  // Converts a Float32Array to base64-encoded PCM16 data
  function base64EncodeAudio(float32Array) {
    const arrayBuffer = floatTo16BitPCM(float32Array);
    let binary = "";
    let bytes = new Uint8Array(arrayBuffer);
    const chunkSize = 0x8000; // 32KB chunk size
    for (let i = 0; i < bytes.length; i += chunkSize) {
      let chunk = bytes.subarray(i, i + chunkSize);
      binary += String.fromCharCode.apply(null, chunk);
    }
    return btoa(binary);
  }

  startButton.onclick = async () => {
    audioContext = new (window.AudioContext || window.webkitAudioContext)({
      sampleRate: 24000,
    });
    const stream = await navigator.mediaDevices.getUserMedia({
      audio: {
        channelCount: 1,
        sampleRate: 24000,
        sampleSize: 16,
      },
    });
    const source = audioContext.createMediaStreamSource(stream);
    const processor = audioContext.createScriptProcessor(1024, 1, 1);

    source.connect(processor);
    processor.connect(audioContext.destination);

    // Send audio chunks continuously as they are processed
    processor.onaudioprocess = (e) => {
      const inputData = e.inputBuffer.getChannelData(0);
      const encodedAudio = base64EncodeAudio(new Float32Array(inputData));
      subscription.sendAudio(encodedAudio); // Send audio data via WebSocket
    };

    startButton.disabled = true;
    stopButton.disabled = false;
  };

  stopButton.onclick = () => {
    audioContext.close();
    startButton.disabled = false;
    stopButton.disabled = true;
    closeSession();
  };
});

open_ai_channel.rb

# app/channels/open_ai_channel.rb
class OpenAiChannel < ApplicationCable::Channel
  def subscribed
    puts "Subscribed to open_ai_#{current_user.id}"
    stream_from "open_ai_#{current_user.id}"
    @openai_client = OpenAiWebsocket.new(current_user.id)
    @openai_client.connect
    @openai_client.session_update(event_id: SecureRandom.uuid)
    item = {
            "id": "msg_001",
            "type": "message",
            "status": "completed",
            "role": "user",
            "content": [
                {
                    "type": "input_text",
                    "text": "You are a representative of Little Bird Electronics.You are on a phone call with a customer from at Little Bird Electronics. IMPORTANT: Keep responses SHORT and curt and polite!!! #{background_info}"
                }
            ]
        }
    @openai_client.conversation_item_create(item: item, event_id: SecureRandom.uuid)
    # @openai_client.create_response(event_id: "12345")
  end

  def background_info
    text = <<~TEXT
      Little Bird Electronics is an Australian electronics retailer.
      We specialize in Arduino, Raspberry Pi, and other maker electronics.
      Our store is located in Sydney, but we ship Australia-wide.
      We offer a wide range of products including sensors, motors, displays, and development boards.
      Our knowledgeable staff can assist with project ideas and technical questions.
      We also run workshops and events for the maker community.

      Question: How can educational institutions and government agencies place purchase orders?
      Answer: Email your purchase order or the list of products you wish to purchase to team@littlebird.com.au.
      Alternatively, place your order online and select the “Purchase Order” option at checkout.
      Question: What payment terms do you offer for educational institutions and government agencies?
      Answer: We offer net payment terms, allowing payment within 30 calendar days from the invoice date.
      Question: How long does it take to process purchase orders?
      Answer: We usually process purchase orders on the same NSW business day, though it might take a bit longer sometimes.
      Question: Where can I find your business details, such as ABN and bank information?
      Answer: Our key business details, including ABN and bank details, are provided below or can be sent upon request.
      Question: How can private businesses place purchase orders?
      Answer: Private businesses can place orders through our website.
      If payment terms are needed, apply before submitting a purchase order.
      Once approved, follow the same process as educational institutions.
      Question: Do private businesses need to apply for payment terms before submitting a purchase order?
      Answer: Yes, private businesses must apply for payment terms prior to submitting a purchase order.
      Question: How can I request a quote for bulk orders or for procedural reasons?
      Answer: Email us at help@littlebird.com.au with the SKU and quantity for each product.
      Include your delivery address and accounts payable information.
      Question: Where can I find the SKU for each product?
      Answer: The SKU is located on the product’s page, just beneath the product title.
      Question: Can I download a copy of my shopping cart?
      Answer: Yes, use the “Download Cart as CSV” button on the cart page.
      Question: Who is eligible for education orders?
      Answer: Public and private schools registered in Australia.
      Australian public and private universities.
      Government-accredited colleges catering to vocational and tertiary education sectors.
      Question: What special pricing do you offer for educational institutions?
      Answer: Special pricing structures for bulk purchases.
      Everyday competitive pricing due to our internet-first business model.
      Question: What is the ordering process for educational institutions?
      Answer: Selection: Browse and select products.
      Request a Quote or Place an Order: Before sending a purchase order, teachers should request a quote or place an order online.
      Prepare Purchase Order: Create a school purchase order or “school order” document with all relevant details.
      Email Order: Send the purchase order to team@littlebird.com.au.
      Confirmation & Processing: Receive order acknowledgment and estimated delivery date.
      Payment & Dispatch: Follow payment instructions; once payment is confirmed, the order will be dispatched.
      Question: Do you offer customization options for education orders?
      Answer: Yes, options include tailored kits and bundles, product modifications, branding and labeling, and special packaging.
      Question: How can I request customization for my order?
      Answer: Indicate your interest in customization when placing an order or requesting a quote.
      Our team will contact you to discuss specific needs.
      Question: What are your payment terms for education orders?
      Answer: Payment is due within 30 calendar days from the invoice date.
      These terms are firm and non-negotiable.
      Question: What payment methods are accepted?
      Answer: Various payment methods are accepted; details are provided on the invoice.
      Question: How do you handle shipping and delivery for education orders?
      Answer: Carrier Partner: Australia Post.
      Signature on Delivery: Required for all orders.
      Arrangements During School Holidays: Deliveries can be scheduled to avoid school closures.
      Tracking: A tracking number is provided once the order is dispatched.
      Question: What is your returns and warranty policy for education orders?
      Answer: No Returns for Change of Mind: Returns are not accepted for unused products or change of mind.
      Faulty Products: Returns are accepted for products with genuine faults.
      Consumables: Excluded from returns and warranty policy.
      Initiating a Return: Contact us with order details and a description of the issue.
      Question: What support and resources do you offer for education orders?
      Answer: Dedicated Support Team: Available to address queries and technical issues.
      Online Resources: User manuals, how-to guides, tutorial videos, and FAQs.
      Training and Workshops: Offered periodically.
      Community Engagement: Opportunities to collaborate with other educators.
      Question: How can I contact Little Bird Electronics?
      Answer: Phone: 1300 240 817
      Email: team@littlebird.com.au
      Physical Address: Unit 13, 8-12 Leighton Place, Hornsby, NSW 2077, Australia
      Mailing Address: PO Box 5036, South Turramurra, NSW 2074, Australia
      Question: What are your key business details?
      Answer: Business Name: Little Bird Electronics Pty Ltd
      ABN: 15 634 521 449
      ACN: 634 521 449
      Question: Who should I contact for accounts and remittance inquiries?
      Answer: Contact Person: Madeleine Schappi
      Email: team@littlebird.com.au
      Phone: 1300 240 817
      Question: What are your Supplier IDs for educational departments?
      Answer: NSW DET Vendor ID: 100135487
      QLD OneSchool Supplier Number: S20017189
      SA Dept. of Education Supplier ID: LBE2077
      Question: Do you have insurance coverage details available?
      Answer: Yes, we have Workers Compensation, Public Liability, Product Liability, and Professional Indemnity insurance. Certificates of Currency can be provided upon request.
      Question: What should I do if I need assistance during the ordering process?
      Answer: Contact our dedicated team at team@littlebird.com.au or call 1300 240 817 for support.
      Question: Can I arrange special packaging or branding for bulk educational orders?
      Answer: Yes, we offer branding and special packaging options for bulk orders. Please discuss your requirements with our team.
      Question: How can I track the status of my order?
      Answer: A tracking number will be emailed to you once your order has been dispatched, allowing you to monitor delivery progress.
      Question: Are there any resources to help integrate your products into our curriculum?
      Answer: Yes, we offer online resources and can provide training or workshops to help educators effectively use our products.
      Question: What is the procedure if I receive a faulty product?
      Answer: Contact us immediately with your order number and details of the fault. We will guide you through the return and replacement process.
      Question: Do you offer discounts for bulk purchases?
      Answer: Yes, we offer special pricing structures for bulk orders by educational institutions.
      Question: Is it possible to get an official quote for procurement purposes?
      Answer: Absolutely, email us the SKUs and quantities needed along with your delivery and accounts payable information to receive an official quote.
      Question: Do you require verification for educational institution orders?
      Answer: Verification may be required. An official representative must complete the ordering process, and relevant documentation may be requested.
      Question: Can I arrange for my order to be delivered after a specific date?
      Answer: Yes, please inform us of any delivery scheduling needs, and we will accommodate them accordingly.
      Question: Are there any exclusions to your returns policy?
      Answer: Yes, consumable items are excluded, and we do not accept returns for change of mind or unused products.
      Question: How often do you update your product offerings?
      Answer: We regularly update our product range and will keep you informed about new releases and enhancements.
      Question: What is the best way to stay informed about your latest products and offers?
      Answer: Join our mailing list or follow us on social media for updates on products, offers, and educational resources.
      Question: Do you provide support after the purchase?
      Answer: Yes, our support doesn’t end after the sale. We’re here to assist with any questions or issues you may have with our products.
      Question: Can I get a copy of your Certificates of Currency for insurance purposes?
      Answer: Yes, we can provide Certificates of Currency for our insurance policies upon request.
      Question: What are your office hours?
      Answer: Our office hours are standard NSW business hours. Please contact us for specific times.
      Question: Do you have a fax number?
      Answer: Yes, our fax number is (02) 8319 2017.
      Question: Can I apply for payment terms as a private business?
      Answer: Yes, private businesses can apply for payment terms, but approval is required before submitting a purchase order.
      Question: Do you offer international shipping?
      Answer: The provided information focuses on Australian educational institutions. Please contact us directly to inquire about international shipping options.
      Question: What is your policy on product warranties?
      Answer: Warranty durations vary by product. Specific warranty information is provided with the product or upon request.
      Question: How do I become part of your educator community?
      Answer: Reach out to us at team@littlebird.com.au to learn more about joining our community of educators.
      Question: Are there any special considerations for orders during public holidays?
      Answer: Yes, processing and delivery times may be affected during public holidays. Please contact us to make any necessary arrangements.
      Question: Do you provide invoicing options compatible with our procurement systems?
      Answer: Yes, we can accommodate various invoicing requirements. Please provide us with the necessary specifications.
      Question: How can I ensure my institution gets the best possible pricing?
      Answer: For bulk orders or special pricing inquiries, contact our sales team to discuss available discounts and offers.
      Question: Is technical support available for your products?
      Answer: Yes, technical support is available through our dedicated support team and online resources.
      Question: What steps do you take to ensure product quality?
      Answer: We adhere to strict quality control measures and partner with reputable manufacturers to ensure high-quality products.
      Question: Can I schedule a demonstration of your products?
      Answer: Please contact us to discuss the possibility of product demonstrations or presentations for your institution.

    TEXT

    text
  end

  def unsubscribed
    @openai_client.close if @openai_client
  end

  def append_audio(data)
    @openai_client.append_audio(data["audio"])
  end

  def close_session
    @openai_client.close
  end

  def restart_session
    @openai_client = OpenAiWebsocket.new(current_user.id)
    @openai_client.connect
  end

  def cancel_response
    @openai_client.cancel_response
  end

  def receive(data)
    case data["type"]
    when "audio"
      @openai_client.input_audio_buffer_append(data["audio"])
    when "text"
      @openai_client.add_message(data["text"])
    end
  end
end

open_ai_websocket.rb

# app/services/open_ai_websocket.rb
class OpenAiWebsocket
  URL = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"

  def initialize(session_id)
    @api_key = Rails.application.credentials.open_ai[:api_key]
    @session_id = session_id
    @ws = nil
    @message_queue = Queue.new
    @connected = false
  end

  def headers
    {
      "Authorization" => "Bearer #{@api_key}",
      "OpenAI-Beta" => "realtime=v1"
    }
  end

  def connect
    Thread.new { run_eventmachine }
    wait_for_connection
  end

  def run_eventmachine
    EM.run do
      puts "Connecting to OpenAI WebSocket..."
      @ws = Faye::WebSocket::Client.new(URL, nil, headers:)
      puts "DateTime: #{DateTime.now}"
      setup_response_handlers
      setup_event_handlers

      EM.add_periodic_timer(0.1) { process_message_queue }
    end
  end

  def wait_for_connection
    sleep 0.1 until @connected
  end

  def process_message_queue
    until @message_queue.empty?
      message = @message_queue.pop
      send_message(message)
    end
  end

  def enqueue_message(message)
    @message_queue.push(message)
  end

  def session_update(event_id:)
    enqueue_message({
      event_id: event_id,
      type: "session.update",
      session: {
        turn_detection: {
          type: "server_vad",
          threshold: 0.5,
          prefix_padding_ms: 300,
          silence_duration_ms: 200
        }
      }
    })
  end

  alias_method :update_session, :session_update

  def input_audio_buffer_append(audio_data)
    enqueue_message({
      event_id: "event_456",
      type: "input_audio_buffer.append",
      audio: audio_data
    })
  end

  alias_method :append_audio_buffer, :input_audio_buffer_append

  def input_audio_buffer_commit(event_id: nil)
    enqueue_message({
      event_id: event_id,
      type: "input_audio_buffer.commit"
    })
  end

  alias_method :commit_audio_buffer, :input_audio_buffer_commit

  def input_audio_buffer_clear(event_id: nil)
    enqueue_message({
      event_id: event_id,
      type: "input_audio_buffer.clear"
    })
  end

  alias_method :clear_audio_buffer, :input_audio_buffer_clear

  def conversation_item_create(item:, event_id: nil, previous_item_id: nil)
    enqueue_message({
      event_id: event_id,
      type: "conversation.item.create",
      previous_item_id: previous_item_id,
      item: item
    })
  end

  alias_method :create_conversation_item, :conversation_item_create

  def add_message(text, event_id: nil, previous_item_id: nil)
    conversation_item_create(
      item: {
        id: event_id,
        type: "message",
        status: "completed",
        role: "user",
        content: [
          {
            type: "input_text",
            text: text
          }
        ]
      },
      event_id:,
      previous_item_id:
    )
  end

  def conversation_item_truncate(item_id:, event_id: nil, content_index: 0, audio_end_ms: 1500)
    enqueue_message({
      event_id: event_id,
      type: "conversation.item.truncate",
      item_id: item_id,
      content_index: content_index,
      audio_end_ms: audio_end_ms
    })
  end

  alias_method :truncate_conversation_item, :conversation_item_truncate

  def conversation_item_delete(event_id:, item_id:)
    enqueue_message({
      event_id: event_id,
      type: "conversation.item.delete",
      item_id: item_id
    })
  end

  alias_method :delete_conversation_item, :conversation_item_delete

  def response_create(event_id: nil)
    enqueue_message({
      event_id: event_id,
      type: "response.create",
      response: {
        modalities: ["text", "audio"],
        instructions: "Please assist the user.  John Croucher is your awesome cofounder at Chick Commerce",
        voice: "alloy",
        output_audio_format: "pcm16",
        tools: [],
        tool_choice: "auto",
        temperature: 0.7,
        max_output_tokens: 150
      }
    })
  end

  alias_method :create_response, :response_create

  def response_cancel(event_id)
    enqueue_message({
      event_id: "#{event_id}",
      type: "response.cancel"
    })
  end

  alias_method :cancel_response, :response_cancel

  def append_audio(audio_data)
    enqueue_message({
      type: "input_audio_buffer.append",
      audio: audio_data
    })
  end

  def commit_audio
    enqueue_message({type: "input_audio_buffer.commit"})
  end

  def close
    @ws&.close
  end

  def response_handlers
    {
      :error => :handle_response_error,
      :"session.created" => :handle_session_created,
      :"session.updated" => :handle_session_updated,
      :"conversation.created" => :handle_conversation_created,
      :"input_audio_buffer.committed" => :handle_input_audio_buffer_committed,
      :"input_audio_buffer.cleared" => :handle_input_audio_buffer_cleared,
      :"input_audio_buffer.speech_started" => :handle_input_audio_buffer_speech_started,
      :"input_audio_buffer.speech_stopped" => :handle_input_audio_buffer_speech_stopped,
      :"conversation.item.created" => :handle_conversation_item_created,
      :"conversation.item.input_audio_transcription.completed" => :handle_conversation_item_input_audio_transcription_completed,
      :"conversation.item.input_audio_transcription.failed" => :handle_conversation_item_input_audio_transcription_failed,
      :"conversation.item.truncated" => :handle_conversation_item_truncated,
      :"conversation.item.deleted" => :handle_conversation_item_deleted,
      :"response.created" => :handle_response_created,
      :"response.done" => :handle_response_done,
      :"response.output_item.added" => :handle_response_output_item_added,
      :"response.output_item.done" => :handle_response_output_item_done,
      :"response.content_part.added" => :handle_response_content_part_added,
      :"response.content_part.done" => :handle_response_content_part_done,
      :"response.text.delta" => :handle_response_text_delta,
      :"response.text.done" => :handle_response_text_done,
      :"response.audio_transcript.delta" => :handle_response_audio_transcript_delta,
      :"response.audio_transcript.done" => :handle_response_audio_transcript_done,
      :"response.audio.delta" => :handle_response_audio_delta,
      :"response.audio.done" => :handle_response_audio_done,
      :"response.function_call_arguments.delta" => :handle_response_function_call_arguments_delta,
      :"response.function_call_arguments.done" => :handle_response_function_call_arguments_done,
      :"rate_limits.updated" => :handle_rate_limits_updated
    }
  end

  private

  def setup_event_handlers
    @ws.on(:open) do |_event|
      puts "Connected to OpenAI WebSocket server for session #{@session_id}."
      @connected = true
    end

    @ws.on(:message) do |event|
      message = JSON.parse(event.data)
      handle_message(message)
    end

    @ws.on(:close) do |event|
      puts "Connection closed for session #{@session_id}, code: #{event.code}, reason: #{event.reason}"
      @connected = false
      EM.stop
    end

    @ws.on(:error) do |error|
      puts "WebSocket Error for session #{@session_id}: #{error.message}"
    end
  end

  def setup_response_handlers
    response_handlers.each do |event, handler|
      puts "Setting up response handler for #{event}"
      @ws.on(event, &method(handler))
    end
  end

  def handle_message(message)
    if message["type"] == "input_audio_buffer.speech_started"
      handle_input_audio_buffer_speech_started(message)
    elsif message["type"] == "response.audio.delta"
      broadcast_audio_delta(message["delta"])
    else
      log_message(message)
    end
  end

  def log_message(message)
    puts "Received message for session #{@session_id}: #{message}"
  end

  def send_message(message)
    if @ws && @ws.ready_state == Faye::WebSocket::API::OPEN
      @ws.send(message.to_json)
    else
      puts "WebSocket not ready for session #{@session_id}. Message not sent: #{message}"
    end
  end

  # Server Event Handlers
  def handle_response_error(message)
    puts "Error: #{message}"
  end

  def handle_session_created(message)
    puts "Session Created: #{message}"
  end

  def handle_session_updated(message)
    puts "Session Updated: #{message}"
  end

  def handle_conversation_created(message)
    puts "Conversation Created: #{message}"
  end

  def handle_input_audio_buffer_committed(message)
    puts "Input Audio Buffer Committed: #{message}"
  end

  def handle_input_audio_buffer_cleared(message)
    puts "Input Audio Buffer Cleared: #{message}"
  end

  def handle_input_audio_buffer_speech_started(message)
    puts "Input Audio Buffer Speech Started: #{message}"
    ActionCable.server.broadcast("open_ai_#{@session_id}", {type: "input_audio_buffer.speech_started", message: message})
  end

  def handle_input_audio_buffer_speech_stopped(message)
    puts "Input Audio Buffer Speech Stopped: #{message}"
  end

  def handle_conversation_item_created(message)
    puts "Conversation Item Created: #{message}"
  end

  def handle_conversation_item_input_audio_transcription_completed(message)
    puts "Conversation Item Input Audio Transcription Completed: #{message}"
  end

  def handle_conversation_item_input_audio_transcription_failed(message)
    puts "Conversation Item Input Audio Transcription Failed: #{message}"
  end

  def handle_conversation_item_truncated(message)
    puts "Conversation Item Truncated: #{message}"
  end

  def handle_conversation_item_deleted(message)
    puts "Conversation Item Deleted: #{message}"
  end

  def handle_response_created(message)
    puts "Response Created: #{message}"
  end

  def handle_response_done(message)
    puts "Response Done: #{message}"
  end

  def handle_response_output_item_added(message)
    puts "Response Output Item Added: #{message}"
  end

  def handle_response_output_item_done(message)
    puts "Response Output Item Done: #{message}"
  end

  def handle_response_content_part_added(message)
    puts "Response Content Part Added: #{message}"
  end

  def handle_response_content_part_done(message)
    puts "Response Content Part Done: #{message}"
  end

  def handle_response_text_done(message)
    puts "Response Text Done: #{message}"
  end

  def handle_response_audio_transcript_delta(message)
    puts "Response Audio Transcript Delta: #{message}"
  end

  def handle_response_audio_transcript_done(message)
    puts "Response Audio Transcript Done: #{message}"
  end

  def handle_response_audio_delta(message)
    # puts "Response Audio Delta: #{message}"
    ActionCable.server.broadcast("open_ai_#{@session_id}", {audio_delta: message["delta"]})
  end

  def handle_response_audio_done(message)
    puts "Response Audio Done: #{message}"
  end

  def handle_response_function_call_arguments_delta(message)
    puts "Response Function Call Arguments Delta: #{message}"
  end

  def handle_response_function_call_arguments_done(message)
    puts "Response Function Call Arguments Done: #{message}"
  end

  def handle_rate_limits_updated(message)
    puts "Rate Limits Updated: #{message}"
  end

  def broadcast_audio_delta(delta)
    ActionCable.server.broadcast("open_ai_#{@session_id}", {type: "audio", data: delta})
  end

  # Modify other broadcast methods similarly
  def handle_response_text_delta(message)
    ActionCable.server.broadcast("open_ai_#{@session_id}", {type: "text", data: message})
  end
end

Thanks for the example ❤️
Have you considered updating this to use audioWorklets? I get deprecation errors on chrome

I wasn’t aware of AudioWorklets; I’ll need to check them out. Thanks for the heads-up!

Hi @schappim , thanks for the reply, also could I suggest to change the Thread.new to use Async ? I believe the Ruby Fibers has less memory cost than a Thread so in theor´y we could spin more connections

   require "async"

  def connect
    Async do |task|
      task.async do
        run_eventmachine
      end
    end
  end

and then, in order to update the session instead of waiting for @connected == true, we could react on the session.created like:

def handle_message(message)
    if message["type"] == "input_audio_buffer.speech_started"
      handle_input_audio_buffer_speech_started(message)
    elsif message["type"] == "session.created"
      session_update(event_id: SecureRandom.uuid)
      ....

Thank you for this, it's great! Maybe someone worked out how to record the audio from both streams too ?